[SPARKR-153] phase 1: implement fold() and aggregate().

Author: Sun Rui

pkg/NAMESPACE

@@ -2,6 +2,7 @@
 exportClasses("RDD")
 exportClasses("Broadcast")
 exportMethods(
+              "aggregateRDD",
               "cache",
               "checkpoint",
               "cogroup",
@@ -17,6 +18,7 @@ exportMethods(
               "filterRDD",
               "flatMap",
               "flatMapValues",
+              "fold",
               "foreach",
               "foreachPartition",
               "fullOuterJoin",

pkg/R/RDD.R

@@ -1374,6 +1374,71 @@ setMethod("top",
             takeOrderedElem(rdd, num, FALSE)
           })
 
+#' Fold an RDD using a given associative function and a neutral "zero value".
+#'
+#' Aggregate the elements of each partition, and then the results for all the
+#' partitions, using a given associative function and a neutral "zero value".
+#' 
+#' @param rdd An RDD.
+#' @param zeroValue A neutral "zero value".
+#' @param op An associative function for the folding operation.
+#' @return The folding result.
+#' @rdname fold
+#' @export
+#' @examples
+#'\dontrun{
+#' sc <- sparkR.init()
+#' rdd <- parallelize(sc, list(1, 2, 3, 4, 5))
+#' fold(rdd, 0, "+") # 15
+#'}
+setGeneric("fold", function(rdd, zeroValue, op) { standardGeneric("fold") })
+
+#' @rdname fold
+#' @aliases fold,RDD,RDD-method
+setMethod("fold",
+          signature(rdd = "RDD", zeroValue = "ANY", op = "ANY"),
+          function(rdd, zeroValue, op) {
+            aggregateRDD(rdd, zeroValue, op, op)
+          })
+
+#' Aggregate an RDD using the given combine functions and a neutral "zero value".
+#'
+#' Aggregate the elements of each partition, and then the results for all the
+#' partitions, using given combine functions and a neutral "zero value".
+#' 
+#' @param rdd An RDD.
+#' @param zeroValue A neutral "zero value".
+#' @param seqOp A function to aggregate the RDD elements. It may return a different
+#'              result type from the type of the RDD elements.
+#' @param combOp A function to aggregate results of seqOp.
+#' @return The aggregation result.
+#' @rdname aggregateRDD
+#' @export
+#' @examples
+#'\dontrun{
+#' sc <- sparkR.init()
+#' rdd <- parallelize(sc, list(1, 2, 3, 4))
+#' zeroValue <- list(0, 0)
+#' seqOp <- function(x, y) { list(x[[1]] + y, x[[2]] + 1) }
+#' combOp <- function(x, y) { list(x[[1]] + y[[1]], x[[2]] + y[[2]]) }
+#' aggregateRDD(rdd, zeroValue, seqOp, combOp) # list(10, 4)
+#'}
+setGeneric("aggregateRDD", function(rdd, zeroValue, seqOp, combOp) { standardGeneric("aggregateRDD") })
+
+#' @rdname aggregateRDD
+#' @aliases aggregateRDD,RDD,RDD-method
+setMethod("aggregateRDD",
+          signature(rdd = "RDD", zeroValue = "ANY", seqOp = "ANY", combOp = "ANY"),
+          function(rdd, zeroValue, seqOp, combOp) {        
+            partitionFunc <- function(part) {
+              Reduce(seqOp, part, zeroValue)
+            }
+            
+            partitionList <- collect(lapplyPartition(rdd, partitionFunc),
+                                     flatten = FALSE)
+            Reduce(combOp, partitionList, zeroValue)
+          })
+
 ############ Shuffle Functions ############
 
 #' Partition an RDD by key

pkg/inst/tests/test_rdd.R

@@ -302,6 +302,28 @@ test_that("top() on RDDs", {
   expect_equal(actual, as.list(sort(unlist(l), decreasing = TRUE))[1:3])
 })
 
+test_that("fold() on RDDs", {
+  actual <- fold(rdd, 0, "+")
+  expect_equal(actual, Reduce("+", nums, 0))
+  
+  rdd <- parallelize(sc, list())
+  actual <- fold(rdd, 0, "+")
+  expect_equal(actual, 0)
+})
+
+test_that("aggregate() on RDDs", {
+  rdd <- parallelize(sc, list(1, 2, 3, 4))
+  zeroValue <- list(0, 0)
+  seqOp <- function(x, y) { list(x[[1]] + y, x[[2]] + 1) }
+  combOp <- function(x, y) { list(x[[1]] + y[[1]], x[[2]] + y[[2]]) }
+  actual <- aggregateRDD(rdd, zeroValue, seqOp, combOp)
+  expect_equal(actual, list(10, 4))
+  
+  rdd <- parallelize(sc, list())
+  actual <- aggregateRDD(rdd, zeroValue, seqOp, combOp)
+  expect_equal(actual, list(0, 0))
+})
+
 test_that("keys() on RDDs", {
   keys <- keys(intRdd)
   actual <- collect(keys)

pkg/man/aggregateRDD.Rd

@@ -0,0 +1,40 @@
+% Generated by roxygen2 (4.0.2): do not edit by hand
+\docType{methods}
+\name{aggregateRDD}
+\alias{aggregateRDD}
+\alias{aggregateRDD,RDD,RDD-method}
+\alias{aggregateRDD,RDD-method}
+\title{Aggregate an RDD using the given combine functions and a neutral "zero value".}
+\usage{
+aggregateRDD(rdd, zeroValue, seqOp, combOp)
+
+\S4method{aggregateRDD}{RDD}(rdd, zeroValue, seqOp, combOp)
+}
+\arguments{
+\item{rdd}{An RDD.}
+
+\item{zeroValue}{A neutral "zero value".}
+
+\item{seqOp}{A function to aggregate the RDD elements. It may return a different
+result type from the type of the RDD elements.}
+
+\item{combOp}{A function to aggregate results of seqOp.}
+}
+\value{
+The aggregation result.
+}
+\description{
+Aggregate the elements of each partition, and then the results for all the
+partitions, using given combine functions and a neutral "zero value".
+}
+\examples{
+\dontrun{
+sc <- sparkR.init()
+rdd <- parallelize(sc, list(1, 2, 3, 4))
+zeroValue <- list(0, 0)
+seqOp <- function(x, y) { list(x[[1]] + y, x[[2]] + 1) }
+combOp <- function(x, y) { list(x[[1]] + y[[1]], x[[2]] + y[[2]]) }
+aggregateRDD(rdd, zeroValue, seqOp, combOp) # list(10, 4)
+}
+}
+

pkg/man/fold.Rd

@@ -0,0 +1,34 @@
+% Generated by roxygen2 (4.0.2): do not edit by hand
+\docType{methods}
+\name{fold}
+\alias{fold}
+\alias{fold,RDD,RDD-method}
+\alias{fold,RDD-method}
+\title{Fold an RDD using a given associative function and a neutral "zero value".}
+\usage{
+fold(rdd, zeroValue, op)
+
+\S4method{fold}{RDD}(rdd, zeroValue, op)
+}
+\arguments{
+\item{rdd}{An RDD.}
+
+\item{zeroValue}{A neutral "zero value".}
+
+\item{op}{An associative function for the folding operation.}
+}
+\value{
+The folding result.
+}
+\description{
+Aggregate the elements of each partition, and then the results for all the
+partitions, using a given associative function and a neutral "zero value".
+}
+\examples{
+\dontrun{
+sc <- sparkR.init()
+rdd <- parallelize(sc, list(1, 2, 3, 4, 5))
+fold(rdd, 0, "+") # 15
+}
+}
+