[SPARK-7264][ML] Parallel lapply for sparkR

R/pkg/NAMESPACE

@@ -295,6 +295,7 @@ export("as.DataFrame",
        "read.json",
        "read.parquet",
        "read.text",
+       "spark.lapply",
        "sql",
        "str",
        "tableToDF",

R/pkg/R/context.R

@@ -226,6 +226,49 @@ setCheckpointDir <- function(sc, dirName) {
   invisible(callJMethod(sc, "setCheckpointDir", suppressWarnings(normalizePath(dirName))))
 }
 
+#' @title Run a function over a list of elements, distributing the computations with Spark.
+#'
+#' @description
+#' Applies a function in a manner that is similar to doParallel or lapply to elements of a list.
+#' The computations are distributed using Spark. It is conceptually the same as the following code:
+#'   lapply(list, func)
+#'
+#' Known limitations:
+#'  - variable scoping and capture: compared to R's rich support for variable resolutions, the
+# distributed nature of SparkR limits how variables are resolved at runtime. All the variables
+# that are available through lexical scoping are embedded in the closure of the function and
+# available as read-only variables within the function. The environment variables should be
+# stored into temporary variables outside the function, and not directly accessed within the
+# function.
+#'
+#'  - loading external packages: In order to use a package, you need to load it inside the
+#'    closure. For example, if you rely on the MASS module, here is how you would use it:
+#'
+#'\dontrun{
+#' train <- function(hyperparam) {
+#'   library(MASS)
+#'   lm.ridge(“y ~ x+z”, data, lambda=hyperparam)
+#'   model
+#' }
+#'}
+#'
+#' @rdname spark.lapply
+#' @param list the list of elements
+#' @param func a function that takes one argument.
+#' @return a list of results (the exact type being determined by the function)
+#' @export
+#' @examples
+#'\dontrun{
+#' doubled <- spark.lapply(1:10, function(x){2 * x})
+#'}
+spark.lapply <- function(list, func) {
+  sc <- get(".sparkRjsc", envir = .sparkREnv)
+  rdd <- parallelize(sc, list, length(list))
+  results <- map(rdd, func)
+  local <- collect(results)
+  local
+}
+
 #' Set new log level
 #'
 #' Set new log level: "ALL", "DEBUG", "ERROR", "FATAL", "INFO", "OFF", "TRACE", "WARN"

R/pkg/inst/tests/testthat/test_context.R

@@ -141,3 +141,8 @@ test_that("sparkJars sparkPackages as comma-separated strings", {
   expect_that(processSparkJars(f), not(gives_warning()))
   expect_match(processSparkJars(f), f)
 })
+
+test_that("sparkLapply should perform simple transforms", {
+  doubled <- spark.lapply(1:10, function(x){2 * x})
+  expect_equal(doubled, as.list(2 * 1:10))
+})