Merge pull request apache#131 from shivaram/rJavaExpt

[SPARKR-177] New R to Java bridge

Author: shivaram

.gitignore

@@ -18,3 +18,5 @@ work/
 *~
 .Rproj.user
 SparkR-pkg.Rproj
+*.o
+*.so

pkg/DESCRIPTION

@@ -5,10 +5,10 @@ Version: 0.1
 Date: 2013-09-09
 Author: Shivaram Venkataraman
 Maintainer: Shivaram Venkataraman <[email protected]>
+Imports: methods
 Depends:
     R (>= 3.0),
     methods,
-    rJava
 Suggests:
     testthat
 Description: R frontend for Spark

pkg/NAMESPACE

@@ -65,4 +65,5 @@ export(
        "setCheckpointDir"
       )
 export("sparkR.init")
+export("print.jobj")
 useDynLib(SparkR, stringHashCode)

pkg/R/RDD.R

@@ -42,8 +42,8 @@ setGeneric("value", function(bcast) { standardGeneric("value") })
 setMethod("value",
           signature(bcast = "Broadcast"),
           function(bcast) {
-            if (exists(bcast@id, envir=.broadcastValues)) {
-              get(bcast@id, envir=.broadcastValues)
+            if (exists(bcast@id, envir = .broadcastValues)) {
+              get(bcast@id, envir = .broadcastValues)
             } else {
               NULL
             }

pkg/R/broadcast.R

@@ -2,8 +2,7 @@
 
 getMinSplits <- function(sc, minSplits) {
   if (is.null(minSplits)) {
-    ssc <- .jcall(sc, "Lorg/apache/spark/SparkContext;", "sc")
-    defaultParallelism <- .jcall(ssc, "I", "defaultParallelism")
+    defaultParallelism <- callJMethod(sc, "defaultParallelism")
     minSplits <- min(defaultParallelism, 2)
   }
   as.integer(minSplits)
@@ -30,10 +29,9 @@ textFile <- function(sc, path, minSplits = NULL) {
   # Allow the user to have a more flexible definiton of the text file path
   path <- normalizePath(path)
   #' Convert a string vector of paths to a string containing comma separated paths
-  path <- paste(path, collapse=",")
+  path <- paste(path, collapse = ",")
 
-  jrdd <- .jcall(sc, "Lorg/apache/spark/api/java/JavaRDD;", "textFile", path,
-                 getMinSplits(sc, minSplits))
+  jrdd <- callJMethod(sc, "textFile", path, getMinSplits(sc, minSplits))
   RDD(jrdd, FALSE)
 }
 
@@ -58,10 +56,9 @@ objectFile <- function(sc, path, minSplits = NULL) {
   # Allow the user to have a more flexible definiton of the text file path
   path <- normalizePath(path)
   #' Convert a string vector of paths to a string containing comma separated paths
-  path <- paste(path, collapse=",")
+  path <- paste(path, collapse = ",")
 
-  jrdd <- .jcall(sc, "Lorg/apache/spark/api/java/JavaRDD;", "objectFile", path,
-                 getMinSplits(sc, minSplits))
+  jrdd <- callJMethod(sc, "objectFile", path, getMinSplits(sc, minSplits))
   # Assume the RDD contains serialized R objects.
   RDD(jrdd, TRUE)
 }
@@ -106,16 +103,8 @@ parallelize <- function(sc, coll, numSlices = 1) {
   # 2-tuples of raws
   serializedSlices <- lapply(slices, serialize, connection = NULL)
 
-  javaSerializedSlices <- .jarray(lapply(serializedSlices, .jarray),
-                                  contents.class = "[B")
-
-  jrddType = "Lorg/apache/spark/api/java/JavaRDD;"
-
-  jrdd <- .jcall("edu/berkeley/cs/amplab/sparkr/RRDD",
-                 jrddType,
-                 "createRDDFromArray",
-                 sc,
-                 javaSerializedSlices)
+  jrdd <- callJStatic("edu.berkeley.cs.amplab.sparkr.RRDD",
+                      "createRDDFromArray", sc, serializedSlices)
 
   RDD(jrdd, TRUE)
 }
@@ -186,12 +175,10 @@ includePackage <- function(sc, pkg) {
 broadcast <- function(sc, object) {
   objName <- as.character(substitute(object))
   serializedObj <- serialize(object, connection = NULL, ascii = TRUE)
-  serializedObjArr <- .jcast(.jarray(serializedObj),
-                             new.class="java/lang/Object")
-  jBroadcast <- .jcall(sc, "Lorg/apache/spark/broadcast/Broadcast;",
-                       "broadcast", serializedObjArr)
 
-  id <- as.character(.jsimplify(.jcall(jBroadcast, "J", "id")))
+  jBroadcast <- callJMethod(sc, "broadcast", serializedObj)
+  id <- as.character(callJMethod(jBroadcast, "id"))
+
   Broadcast(id, object, jBroadcast, objName)
 }
 
@@ -211,8 +198,5 @@ broadcast <- function(sc, object) {
 #' checkpoint(rdd)
 #'}
 setCheckpointDir <- function(sc, dirName) {
-  ssc <- .jcall(sc, "Lorg/apache/spark/SparkContext;", "sc")
-  .jcall(ssc, "V", "setCheckpointDir", suppressWarnings(normalizePath(dirName)))
-  # NOTE: rJava doesn't check for exceptions if the return type is void
-  .jcheck()
+  invisible(callJMethod(sc, "setCheckpointDir", suppressWarnings(normalizePath(dirName))))
 }

pkg/R/context.R

@@ -0,0 +1,103 @@
+# Utility functions to deserialize objects from Java.
+
+# Type mapping from Java to R
+# 
+# void -> NULL
+# Int -> integer
+# String -> character
+# Boolean -> logical
+# Double -> double
+# Long -> double
+# Array[Byte] -> raw
+#
+# Array[T] -> list()
+# Object -> jobj
+
+readObject <- function(con) {
+  # Read type first
+  type <- readType(con)
+  readTypedObject(con, type)
+}
+
+readTypedObject <- function(con, type) {
+  switch (type,
+    "i" = readInt(con),
+    "c" = readString(con),
+    "b" = readBoolean(con),
+    "d" = readDouble(con),
+    "r" = readRaw(con),
+    "l" = readList(con),
+    "n" = NULL,
+    "j" = getJobj(readString(con)),
+    stop("Unsupported type for deserialization"))
+}
+
+readString <- function(con) {
+  stringLen <- readInt(con)
+  string <- readBin(con, raw(), stringLen, endian = "big")
+  rawToChar(string)
+}
+
+readInt <- function(con) {
+  readBin(con, integer(), n = 1, endian = "big")
+}
+
+readDouble <- function(con) {
+  readBin(con, double(), n = 1, endian = "big")
+}
+
+readBoolean <- function(con) {
+  as.logical(readInt(con))
+}
+
+readType <- function(con) {
+  rawToChar(readBin(con, "raw", n = 1L))
+}
+
+# We only support lists where all elements are of same type
+readList <- function(con) {
+  type <- readType(con)
+  len <- readInt(con)
+  if (len > 0) {
+    l <- vector("list", len)
+    for (i in 1:len) {
+      l[[i]] <- readTypedObject(con, type)
+    }
+    l
+  } else {
+    list()
+  }
+}
+
+readRaw <- function(con) {
+  dataLen <- readInt(con)
+  data <- readBin(con, raw(), as.integer(dataLen), endian = "big")
+}
+
+readRawLen <- function(con, dataLen) {
+  data <- readBin(con, raw(), as.integer(dataLen), endian = "big")
+}
+
+readDeserialize <- function(con) {
+  # We have two cases that are possible - In one, the entire partition is
+  # encoded as a byte array, so we have only one value to read. If so just
+  # return firstData
+  dataLen <- readInt(con)
+  firstData <- unserialize(
+      readBin(con, raw(), as.integer(dataLen), endian = "big"))
+
+  # Else, read things into a list
+  dataLen <- readInt(con)
+  if (length(dataLen) > 0 && dataLen > 0) {
+    data <- list(firstData)
+    while (length(dataLen) > 0 && dataLen > 0) {
+      data[[length(data) + 1L]] <- unserialize(
+          readBin(con, raw(), as.integer(dataLen), endian = "big"))
+      dataLen <- readInt(con)
+    }
+    unlist(data, recursive = FALSE)
+  } else {
+    firstData
+  }
+}
+

pkg/R/deserialize.R

@@ -0,0 +1,53 @@
+# References to objects that exist on the JVM backend
+# are maintained using the jobj. 
+
+# Maintain a reference count of Java object references
+# This allows us to GC the java object when it is safe
+.validJobjs <- new.env(parent = emptyenv())
+
+# List of object ids to be removed
+.toRemoveJobjs <- new.env(parent = emptyenv())
+
+getJobj <- function(objId) {
+  newObj <- jobj(objId)
+  if (exists(objId, .validJobjs)) {
+    .validJobjs[[objId]] <- .validJobjs[[objId]] + 1
+  } else {
+    .validJobjs[[objId]] <- 1
+  }
+  newObj
+}
+
+# Handler for a java object that exists on the backend.
+jobj <- function(objId) {
+  if (!is.character(objId)) {
+    stop("object id must be a character")
+  }
+  # NOTE: We need a new env for a jobj as we can only register
+  # finalizers for environments or external references pointers.
+  obj <- structure(new.env(parent = emptyenv()), class = "jobj")
+  obj$id <- objId
+  # Register a finalizer to remove the Java object when this reference
+  # is garbage collected in R
+  reg.finalizer(obj, cleanup.jobj)
+  obj
+}
+
+print.jobj <- function(jobj) {
+  cls <- callJMethod(jobj, "getClass")
+  name <- callJMethod(cls, "getName")
+  cat("Java ref type", name, "id", jobj$id, "\n", sep = " ")
+}
+
+cleanup.jobj <- function(jobj) {
+  objId <- jobj$id
+  .validJobjs[[objId]] <- .validJobjs[[objId]] - 1
+
+  if (.validJobjs[[objId]] == 0) {
+    rm(list = objId, envir = .validJobjs)
+    # NOTE: We cannot call removeJObject here as the finalizer may be run
+    # in the middle of another RPC. Thus we queue up this object Id to be removed
+    # and then run all the removeJObject when the next RPC is called.
+    .toRemoveJobjs[[objId]] <- 1
+  }
+}

pkg/R/jobj.R

@@ -0,0 +1,125 @@
+# Utility functions to serialize R objects so they can be read in Java.
+
+# Type mapping from R to Java
+#  
+# integer -> Int
+# character -> String
+# logical -> Boolean
+# double, numeric -> Double
+# raw -> Array[Byte]
+#
+# list[T] -> Array[T], where T is one of above mentioned types
+# environment -> Map[String, T], where T is a native type
+# jobj -> Object, where jobj is an object created in the backend
+
+writeObject <- function(con, object, writeType = TRUE) {
+  # NOTE: In R vectors have same type as objects. So we don't support
+  # passing in vectors as arrays and instead require arrays to be passed
+  # as lists.
+  if (writeType) {
+    writeType(con, class(object))
+  }
+  switch(class(object),
+         integer = writeInt(con, object),
+         character = writeString(con, object),
+         logical = writeBoolean(con, object),
+         double = writeDouble(con, object),
+         numeric = writeDouble(con, object),
+         raw = writeRaw(con, object),
+         list = writeList(con, object),
+         jobj = writeString(con, object$id),
+         environment = writeEnv(con, object),
+         stop("Unsupported type for serialization"))
+}
+
+writeString <- function(con, value) {
+  writeInt(con, as.integer(nchar(value) + 1))
+  writeBin(value, con, endian = "big")
+}
+
+writeInt <- function(con, value) {
+  writeBin(as.integer(value), con, endian = "big")
+}
+
+writeDouble <- function(con, value) {
+  writeBin(value, con, endian = "big")
+}
+
+writeBoolean <- function(con, value) {
+  # TRUE becomes 1, FALSE becomes 0
+  writeInt(con, as.integer(value))
+}
+
+writeRawSerialize <- function(outputCon, batch) {
+  outputSer <- serialize(batch, ascii = FALSE, conn = NULL)
+  writeRaw(outputCon, outputSer)
+}
+
+writeRaw <- function(con, batch) {
+  writeInt(con, length(batch))
+  writeBin(batch, con, endian = "big")
+}
+
+writeType <- function(con, class) {
+  type <- switch(class,
+                 integer = "i",
+                 character = "c",
+                 logical = "b",
+                 double = "d",
+                 numeric = "d",
+                 raw = "r",
+                 list = "l",
+                 jobj = "j",
+                 environment = "e",
+                 stop("Unsupported type for serialization"))
+  writeBin(charToRaw(type), con)
+}
+
+# Used to pass arrays where all the elements are of the same type
+writeList <- function(con, arr) {
+  # All elements should be of same type
+  elemType <- unique(sapply(arr, function(elem) { class(elem) }))
+  stopifnot(length(elemType) <= 1)
+
+  # TODO: Empty lists are given type "character" right now.
+  # This may not work if the Java side expects array of any other type.
+  if (length(elemType) == 0) {
+    elemType <- class("somestring")
+  }
+
+  writeType(con, elemType)
+  writeInt(con, length(arr))
+
+  if (length(arr) > 0) {
+    for (a in arr) {
+      writeObject(con, a, FALSE)
+    }
+  }
+}
+
+# Used to pass in hash maps required on Java side.
+writeEnv <- function(con, env) {
+  len <- length(env)
+
+  writeInt(con, len)
+  if (len > 0) {
+    writeList(con, as.list(ls(env)))
+    vals <- lapply(ls(env), function(x) { env[[x]] })
+    writeList(con, as.list(vals))
+  }
+}
+
+# Used to serialize in a list of objects where each
+# object can be of a different type. Serialization format is
+# <object type> <object> for each object
+writeArgs <- function(con, args) {
+  if (length(args) > 0) {
+    for (a in args) {
+      writeObject(con, a)
+    }
+  }
+}
+
+writeStrings <- function(con, stringList) {
+  writeLines(unlist(stringList), con)
+}