Merge pull request apache#130 from lythesia/master

[SPARKR-162] add fullOuterJoin in RDD.R

Author: concretevitamin

pkg/NAMESPACE

@@ -19,6 +19,7 @@ exportMethods(
               "flatMapValues",
               "foreach",
               "foreachPartition",
+              "fullOuterJoin",
               "groupByKey",
               "join",
               "keyBy",

pkg/R/RDD.R

@@ -1783,6 +1783,86 @@ setMethod("rightOuterJoin",
             joined <- flatMapValues(groupByKey(unionRDD(rdd1Tagged, rdd2Tagged), numPartitions), doJoin)
           })
 
+#' Full outer join two RDDs
+#'
+#' This function full-outer-joins two RDDs where every element is of the form 
+#' list(K, V). 
+#' The key types of the two RDDs should be the same.
+#'
+#' @param rdd1 An RDD to be joined. Should be an RDD where each element is
+#'             list(K, V).
+#' @param rdd2 An RDD to be joined. Should be an RDD where each element is
+#'             list(K, V).
+#' @param numPartitions Number of partitions to create.
+#' @return For each element (k, v) in rdd1 and (k, w) in rdd2, the resulting RDD
+#'         will contain all pairs (k, (v, w)) for both (k, v) in rdd1 and and
+#'         (k, w) in rdd2, or the pair (k, (NULL, w))/(k, (v, NULL)) if no elements 
+#'         in rdd1/rdd2 have key k.
+#' @rdname fullOuterJoin
+#' @export
+#' @examples
+#'\dontrun{
+#' sc <- sparkR.init()
+#' rdd1 <- parallelize(sc, list(list(1, 2), list(1, 3), list(3, 3)))
+#' rdd2 <- parallelize(sc, list(list(1, 1), list(2, 4)))
+#' fullOuterJoin(rdd1, rdd2, 2L) # list(list(1, list(2, 1)),
+#'                               #      list(1, list(3, 1)),
+#'                               #      list(3, list(3, NULL)),
+#'                               #      list(2, list(NULL, 4)))
+#'}
+setGeneric("fullOuterJoin", function(rdd1, rdd2, numPartitions) { standardGeneric("fullOuterJoin") })
+
+#' @rdname fullOuterJoin
+#' @aliases fullOuterJoin,RDD,RDD-method
+setMethod("fullOuterJoin",
+          signature(rdd1 = "RDD", rdd2 = "RDD", numPartitions = "integer"),
+          function(rdd1, rdd2, numPartitions) {
+            rdd1Tagged <- lapply(rdd1, function(x) { list(x[[1]], list(1L, x[[2]])) })
+            rdd2Tagged <- lapply(rdd2, function(x) { list(x[[1]], list(2L, x[[2]])) })
+
+            doJoin <- function(v) {
+              t1 <- vector("list", length(v))
+              t2 <- vector("list", length(v))
+              index1 <- 1
+              index2 <- 1
+              for (x in v) {
+                if (x[[1]] == 1L) {
+                  t1[[index1]] <- x[[2]]
+                  index1 <- index1 + 1
+                } else {
+                  t2[[index2]] <- x[[2]]
+                  index2 <- index2 + 1
+                }
+              }
+              len1 <- index1 - 1
+              len2 <- index2 - 1
+
+              if (len1 == 0) {
+                t1 <- list(NULL)
+              } else {
+                length(t1) <- len1
+              }
+
+              if (len2 == 0) {
+                t2 <- list(NULL)
+              } else {
+                length(t2) <- len2
+              }
+
+              result <- list()
+              length(result) <- length(t1) * length(t2)
+              index <- 1
+              for(i in t1) {
+                for(j in t2) {
+                  result[[index]] <- list(i, j)
+                  index <- index + 1
+                }
+              }
+              result
+            }
+            joined <- flatMapValues(groupByKey(unionRDD(rdd1Tagged, rdd2Tagged), numPartitions), doJoin)
+          })
+
 #' For each key k in several RDDs, return a resulting RDD that
 #' whose values are a list of values for the key in all RDDs.
 #'

pkg/inst/tests/test_rdd.R

@@ -347,3 +347,29 @@ test_that("rightOuterJoin() on pairwise RDDs", {
                sortKeyValueList(list(list("d", list(NULL, 4)), list("c", list(NULL, 3)))))
 })
 
+test_that("fullOuterJoin() on pairwise RDDs", {
+  rdd1 <- parallelize(sc, list(list(1,2), list(1,3), list(3,3)))
+  rdd2 <- parallelize(sc, list(list(1,1), list(2,4)))
+  actual <- collect(fullOuterJoin(rdd1, rdd2, 2L))
+  expected <- list(list(1, list(2, 1)), list(1, list(3, 1)), list(2, list(NULL, 4)), list(3, list(3, NULL)))
+  expect_equal(sortKeyValueList(actual), sortKeyValueList(expected))
+
+  rdd1 <- parallelize(sc, list(list("a",2), list("a",3), list("c", 1)))
+  rdd2 <- parallelize(sc, list(list("a",1), list("b",4)))
+  actual <- collect(fullOuterJoin(rdd1, rdd2, 2L))
+  expected <- list(list("b", list(NULL, 4)), list("a", list(2, 1)), list("a", list(3, 1)), list("c", list(1, NULL)))
+  expect_equal(sortKeyValueList(actual),
+               sortKeyValueList(expected))
+
+  rdd1 <- parallelize(sc, list(list(1,1), list(2,2)))
+  rdd2 <- parallelize(sc, list(list(3,3), list(4,4)))
+  actual <- collect(fullOuterJoin(rdd1, rdd2, 2L))
+  expect_equal(sortKeyValueList(actual),
+               sortKeyValueList(list(list(1, list(1, NULL)), list(2, list(2, NULL)), list(3, list(NULL, 3)), list(4, list(NULL, 4)))))
+
+  rdd1 <- parallelize(sc, list(list("a",1), list("b",2)))
+  rdd2 <- parallelize(sc, list(list("c",3), list("d",4)))
+  actual <- collect(fullOuterJoin(rdd1, rdd2, 2L))
+  expect_equal(sortKeyValueList(actual),
+               sortKeyValueList(list(list("a", list(1, NULL)), list("b", list(2, NULL)), list("d", list(NULL, 4)), list("c", list(NULL, 3)))))
+})

pkg/man/fullOuterJoin.Rd

@@ -0,0 +1,45 @@
+% Generated by roxygen2 (4.1.0): do not edit by hand
+% Please edit documentation in R/RDD.R
+\docType{methods}
+\name{fullOuterJoin}
+\alias{fullOuterJoin}
+\alias{fullOuterJoin,RDD,RDD,integer-method}
+\alias{fullOuterJoin,RDD,RDD-method}
+\title{Full outer join two RDDs}
+\usage{
+fullOuterJoin(rdd1, rdd2, numPartitions)
+
+\S4method{fullOuterJoin}{RDD,RDD,integer}(rdd1, rdd2, numPartitions)
+}
+\arguments{
+\item{rdd1}{An RDD to be joined. Should be an RDD where each element is
+list(K, V).}
+
+\item{rdd2}{An RDD to be joined. Should be an RDD where each element is
+list(K, V).}
+
+\item{numPartitions}{Number of partitions to create.}
+}
+\value{
+For each element (k, v) in rdd1 and (k, w) in rdd2, the resulting RDD
+        will contain all pairs (k, (v, w)) for both (k, v) in rdd1 and and
+        (k, w) in rdd2, or the pair (k, (NULL, w))/(k, (v, NULL)) if no elements
+        in rdd1/rdd2 have key k.
+}
+\description{
+This function full-outer-joins two RDDs where every element is of the form
+list(K, V).
+The key types of the two RDDs should be the same.
+}
+\examples{
+\dontrun{
+sc <- sparkR.init()
+rdd1 <- parallelize(sc, list(list(1, 2), list(1, 3), list(3, 3)))
+rdd2 <- parallelize(sc, list(list(1, 1), list(2, 4)))
+fullOuterJoin(rdd1, rdd2, 2L) # list(list(1, list(2, 1)),
+                              #      list(1, list(3, 1)),
+                              #      list(3, list(3, NULL)),
+                              #      list(2, list(NULL, 4)))
+}
+}
+