Merge pull request apache#102 from hlin09/hlin09

Adds function flatMapValues.

pkg/NAMESPACE

@@ -13,6 +13,7 @@ exportMethods(
               "Filter",
               "filter",
               "flatMap",
+              "flatMapValues",
               "groupByKey",
               "keys",
               "length",

pkg/R/RDD.R

@@ -426,7 +426,8 @@ setMethod("flatMap",
           function(X, FUN) {
             partitionFunc <- function(part) {
               unlist(
-                lapply(part, FUN)
+                lapply(part, FUN),
+                recursive = F
               )
             }
             lapplyPartition(X, partitionFunc)
@@ -934,6 +935,36 @@ setMethod("mapValues",
             lapply(X, func)
           })
 
+#' Pass each value in the key-value pair RDD through a flatMap function without
+#' changing the keys; this also retains the original RDD's partitioning.
+#' 
+#' The same as 'flatMapValues()' in Spark.
+#'
+#' @param X The RDD to apply the transformation.
+#' @param FUN the transformation to apply on the value of each element.
+#' @return a new RDD created by the transformation.
+#' @rdname flatMapValues
+#' @export
+#' @examples
+#'\dontrun{
+#' sc <- sparkR.init()
+#' rdd <- parallelize(sc, list(list(1, c(1,2)), list(2, c(3,4))))
+#' collect(flatMapValues(rdd, function(x) { x }))
+#' Output: list(list(1,1), list(1,2), list(2,3), list(2,4))
+#'}
+setGeneric("flatMapValues", function(X, FUN) { standardGeneric("flatMapValues") })
+
+#' @rdname flatMapValues
+#' @aliases flatMapValues,RDD,function-method
+setMethod("flatMapValues",
+          signature(X = "RDD", FUN = "function"),
+          function(X, FUN) {
+            flatMapFunc <- function(x) {
+              lapply(FUN(x[[2]]), function(v) { list(x[[1]], v) })
+            }
+            flatMap(X, flatMapFunc)
+          })
+
 ############ Shuffle Functions ############
 
 #' Partition an RDD by key

pkg/inst/tests/test_rdd.R

@@ -33,6 +33,12 @@ test_that("mapPartitions on RDD", {
   expect_equal(actual, list(15, 40))
 })
 
+test_that("flatMap() on RDDs", {
+  flat <- flatMap(intRdd, function(x) { list(x, x) })
+  actual <- collect(flat)
+  expect_equal(actual, rep(intPairs, each=2))
+})
+
 test_that("Filter on RDD", {
   filtered.rdd <- Filter(function(x) { x %% 2 == 0 }, rdd)
   actual <- collect(filtered.rdd)
@@ -186,6 +192,18 @@ test_that("mapValues() on pairwise RDDs", {
     }))
 })
 
+test_that("flatMapValues() on pairwise RDDs", {
+  l <- parallelize(sc, list(list(1, c(1,2)), list(2, c(3,4))))
+  actual <- collect(flatMapValues(l, function(x) { x }))
+  expect_equal(actual, list(list(1,1), list(1,2), list(2,3), list(2,4)))
+
+  # Generate x to x+1 for every value
+  actual <- collect(flatMapValues(intRdd, function(x) { x:(x + 1) }))
+  expect_equal(actual, 
+               list(list(1L, -1), list(1L, 0), list(2L, 100), list(2L, 101),
+                    list(2L, 1), list(2L, 2), list(1L, 200), list(1L, 201)))
+})
+
 test_that("distinct() on RDDs", {
   nums.rep2 <- rep(1:10, 2)
   rdd.rep2 <- parallelize(sc, nums.rep2, 2L)

pkg/man/flatMapValues.Rd

@@ -0,0 +1,32 @@
+% Generated by roxygen2 (4.0.2): do not edit by hand
+\docType{methods}
+\name{flatMapValues}
+\alias{flatMapValues}
+\alias{flatMapValues,RDD,function-method}
+\title{Pass each value in the key-value pair RDD through a flatMap function without
+changing the keys; this also retains the original RDD's partitioning.}
+\usage{
+flatMapValues(X, FUN)
+
+\S4method{flatMapValues}{RDD,`function`}(X, FUN)
+}
+\arguments{
+\item{X}{The RDD to apply the transformation.}
+
+\item{FUN}{the transformation to apply on the value of each element.}
+}
+\value{
+a new RDD created by the transformation.
+}
+\description{
+The same as 'flatMapValues()' in Spark.
+}
+\examples{
+\dontrun{
+sc <- sparkR.init()
+rdd <- parallelize(sc, list(list(1, c(1,2)), list(2, c(3,4))))
+collect(flatMapValues(rdd, function(x) { x }))
+Output: list(list(1,1), list(1,2), list(2,3), list(2,4))
+}
+}
+