Merge pull request apache#120 from sun-rui/objectFile

[SPARKR-146] Support read/save object files in SparkR.

Author: shivaram

pkg/NAMESPACE

@@ -42,6 +42,7 @@ exportMethods(
               "reduceByKey",
               "rightOuterJoin",
               "sampleRDD",
+              "saveAsObjectFile",
               "take",
               "takeSample",
               "unionRDD",
@@ -53,6 +54,7 @@ exportMethods(
 # S3 methods exported
 export(
        "textFile",
+       "objectFile",
        "parallelize",
        "hashCode",
        "includePackage",

pkg/R/RDD.R

@@ -1061,6 +1061,36 @@ setMethod("keyBy",
             lapply(rdd, apply.func)
           })
 
+#' Save this RDD as a SequenceFile of serialized objects.
+#'
+#' @param rdd The RDD to save
+#' @param path The directory where the file is saved
+#' @rdname saveAsObjectFile
+#' @seealso objectFile
+#' @export
+#' @examples
+#'\dontrun{
+#' sc <- sparkR.init()
+#' rdd <- parallelize(sc, 1:3)
+#' saveAsObjectFile(rdd, "/tmp/sparkR-tmp")
+#'}
+setGeneric("saveAsObjectFile", function(rdd, path) { standardGeneric("saveAsObjectFile") })
+
+#' @rdname saveAsObjectFile
+#' @aliases saveAsObjectFile,RDD
+setMethod("saveAsObjectFile",
+          signature(rdd = "RDD", path = "character"),
+          function(rdd, path) {
+            # If the RDD is in string format, need to serialize it before saving it because when
+            # objectFile() is invoked to load the saved file, only serialized format is assumed.
+            if (!rdd@env$serialized) {
+              rdd <- reserialize(rdd)
+            }
+            .jcall(getJRDD(rdd), "V", "saveAsObjectFile", path)
+            # Return nothing
+            invisible(NULL)
+          })
+
 #' Return an RDD with the keys of each tuple.
 #'
 #' @param rdd The RDD from which the keys of each tuple is returned.

pkg/R/context.R

@@ -18,17 +18,46 @@
 #'  lines <- textFile(sc, "myfile.txt")
 #'}
 
-textFile <- function(sc, path, minSplits = NULL) {
+getMinSplits <- function(sc, minSplits) {
   if (is.null(minSplits)) {
     ssc <- .jcall(sc, "Lorg/apache/spark/SparkContext;", "sc")
     defaultParallelism <- .jcall(ssc, "I", "defaultParallelism")
     minSplits <- min(defaultParallelism, 2)
   }
+  as.integer(minSplits)
+}
+
+textFile <- function(sc, path, minSplits = NULL) {
   jrdd <- .jcall(sc, "Lorg/apache/spark/api/java/JavaRDD;", "textFile", path,
-                 as.integer(minSplits))
+                 getMinSplits(sc, minSplits))
   RDD(jrdd, FALSE)
 }
 
+#' Load an RDD saved as a SequenceFile containing serialized objects.
+#'
+#' The file to be loaded should be one that was previously generated by calling
+#' saveAsObjectFile() of the RDD class.
+#'
+#' @param sc SparkContext to use
+#' @param path Path of file to read
+#' @param minSplits Minimum number of splits to be created. If NULL, the default
+#'  value is chosen based on available parallelism.
+#' @return RDD containing serialized R objects.
+#' @seealso saveAsObjectFile
+#' @export
+#' @examples
+#'\dontrun{
+#'  sc <- sparkR.init()
+#'  rdd <- objectFile(sc, "myfile")
+#'}
+
+objectFile <- function(sc, path, minSplits = NULL) {
+  jrdd <- .jcall(sc, "Lorg/apache/spark/api/java/JavaRDD;", "objectFile", path,
+                 getMinSplits(sc, minSplits))
+  # Assume the RDD contains serialized R objects.
+  RDD(jrdd, TRUE)
+}
+
 #' Create an RDD from a homogeneous list or vector.
 #'
 #' This function creates an RDD from a local homogeneous list in R. The elements

pkg/inst/tests/test_binaryFile.R

@@ -0,0 +1,57 @@
+context("functions on binary files")
+
+# JavaSparkContext handle
+sc <- sparkR.init()
+
+mockFile = c("Spark is pretty.", "Spark is awesome.")
+
+test_that("saveAsObjectFile()/objectFile() following textFile() works", {
+  fileName1 <- tempfile(pattern="spark-test", fileext=".tmp")
+  fileName2 <- tempfile(pattern="spark-test", fileext=".tmp")
+  writeLines(mockFile, fileName1)
+
+  rdd <- textFile(sc, fileName1)
+  saveAsObjectFile(rdd, fileName2)
+  rdd <- objectFile(sc, fileName2)
+  expect_equal(collect(rdd), as.list(mockFile))
+
+  unlink(fileName1)
+  unlink(fileName2, recursive = TRUE)
+})
+
+test_that("saveAsObjectFile()/objectFile() works on a parallelized list", {
+  fileName <- tempfile(pattern="spark-test", fileext=".tmp")
+
+  l <- list(1, 2, 3)
+  rdd <- parallelize(sc, l)
+  saveAsObjectFile(rdd, fileName)
+  rdd <- objectFile(sc, fileName)
+  expect_equal(collect(rdd), l)
+
+  unlink(fileName, recursive = TRUE)
+})
+
+test_that("saveAsObjectFile()/objectFile() following RDD transformations works", {
+  fileName1 <- tempfile(pattern="spark-test", fileext=".tmp")
+  fileName2 <- tempfile(pattern="spark-test", fileext=".tmp")
+  writeLines(mockFile, fileName1)
+
+  rdd <- textFile(sc, fileName1)
+
+  words <- flatMap(rdd, function(line) { strsplit(line, " ")[[1]] })
+  wordCount <- lapply(words, function(word) { list(word, 1L) })
+
+  counts <- reduceByKey(wordCount, "+", 2L)
+  
+  saveAsObjectFile(counts, fileName2)
+  counts <- objectFile(sc, fileName2)
+    
+  output <- collect(counts)
+  expected <- list(list("awesome.", 1), list("Spark", 2), list("pretty.", 1),
+                    list("is", 2))
+  expect_equal(output, expected)
+  
+  unlink(fileName1)
+  unlink(fileName2, recursive = TRUE)
+})
+

pkg/man/objectFile.Rd

@@ -0,0 +1,32 @@
+% Generated by roxygen2 (4.0.2): do not edit by hand
+\name{objectFile}
+\alias{objectFile}
+\title{Load an RDD saved as a SequenceFile containing serialized objects.}
+\usage{
+objectFile(sc, path, minSplits = NULL)
+}
+\arguments{
+\item{sc}{SparkContext to use}
+
+\item{path}{Path of file to read}
+
+\item{minSplits}{Minimum number of splits to be created. If NULL, the default
+value is chosen based on available parallelism.}
+}
+\value{
+RDD containing serialized R objects.
+}
+\description{
+The file to be loaded should be one that was previously generated by calling
+saveAsObjectFile() of the RDD class.
+}
+\examples{
+\dontrun{
+ sc <- sparkR.init()
+ rdd <- objectFile(sc, "myfile")
+}
+}
+\seealso{
+saveAsObjectFile
+}
+

pkg/man/saveAsObjectFile.Rd

@@ -0,0 +1,31 @@
+% Generated by roxygen2 (4.0.2): do not edit by hand
+\docType{methods}
+\name{saveAsObjectFile}
+\alias{saveAsObjectFile}
+\alias{saveAsObjectFile,RDD}
+\alias{saveAsObjectFile,RDD,character-method}
+\title{Save this RDD as a SequenceFile of serialized objects.}
+\usage{
+saveAsObjectFile(rdd, path)
+
+\S4method{saveAsObjectFile}{RDD,character}(rdd, path)
+}
+\arguments{
+\item{rdd}{The RDD to save}
+
+\item{path}{The directory where the file is saved}
+}
+\description{
+Save this RDD as a SequenceFile of serialized objects.
+}
+\examples{
+\dontrun{
+sc <- sparkR.init()
+rdd <- parallelize(sc, 1:3)
+saveAsObjectFile(rdd, "/tmp/sparkR-tmp")
+}
+}
+\seealso{
+objectFile
+}
+