Merge pull request apache#90 from oscaroboto/master

Added ExecutorEnv and ability to pass additional JARs to sparkContext (to be used with YARN)

pkg/R/sparkR.R

@@ -7,9 +7,12 @@ sparkR.onLoad <- function(libname, pkgname) {
   packageStartupMessage("[SparkR] Initializing with classpath ", assemblyJarPath, "\n")
 
   sparkMem <- Sys.getenv("SPARK_MEM", "512m")
+  yarn_conf_dir <- Sys.getenv("YARN_CONF_DIR", "")
+
   .sparkREnv$libname <- libname
   .sparkREnv$assemblyJarPath <- assemblyJarPath
   .jinit(classpath=assemblyJarPath, parameters=paste("-Xmx", sparkMem, sep=""))
+  .jaddClassPath(yarn_conf_dir)
 }
 
 #' Initialize a new Spark Context.
@@ -20,19 +23,27 @@ sparkR.onLoad <- function(libname, pkgname) {
 #' @param appName Application name to register with cluster manager
 #' @param sparkHome Spark Home directory
 #' @param sparkEnvir Named list of environment variables to set on worker nodes.
+#' @param sparkExecutorEnv Named list of environment variables to be used when launching executors.
+#' @param sparkJars Character string vector of jar files to pass to the worker nodes.
 #' @export
 #' @examples
 #'\dontrun{
 #' sc <- sparkR.init("local[2]", "SparkR", "/home/spark")
 #' sc <- sparkR.init("local[2]", "SparkR", "/home/spark",
 #'                  list(spark.executor.memory="1g"))
+#' sc <- sparkR.init("yarn-client", "SparkR", "/home/spark",
+#'                  list(spark.executor.memory="1g"),
+#'                  list(LD_LIBRARY_PATH="/directory of Java VM Library Files (libjvm.so) on worker nodes/"),
+#'                  c("jarfile1.jar","jarfile2.jar"))
 #'}
 
 sparkR.init <- function(
   master = "local",
   appName = "SparkR",
   sparkHome = Sys.getenv("SPARK_HOME"),
-  sparkEnvir = list() ) {
+  sparkEnvir = list(),
+  sparkExecutorEnv = list(),
+  sparkJars = "") {
 
   if (exists(".sparkRjsc", envir=.sparkREnv)) {
     return(get(".sparkRjsc", envir=.sparkREnv))
@@ -42,21 +53,32 @@ sparkR.init <- function(
     sparkHome <- normalizePath(sparkHome)
   }
 
-  hm <- .jnew("java/util/HashMap")
+  sparkEnvirMap <- .jnew("java/util/HashMap")
   for (varname in names(sparkEnvir)) {
-    hm$put(varname, sparkEnvir[[varname]])
+    sparkEnvirMap$put(varname, sparkEnvir[[varname]])
   }
-
+
+  sparkExecutorEnvMap <- .jnew("java/util/HashMap")
+  if (!any(names(sparkExecutorEnv) == "LD_LIBRARY_PATH")) {
+    sparkExecutorEnvMap$put("LD_LIBRARY_PATH", paste0("$LD_LIBRARY_PATH:",Sys.getenv("LD_LIBRARY_PATH")))
+  }
+  for (varname in names(sparkExecutorEnv)) {
+    sparkExecutorEnvMap$put(varname, sparkExecutorEnv[[varname]])
+  }
+
+  .jaddClassPath(sparkJars)
+  jars=c(as.character(.sparkREnv$assemblyJarPath), as.character(sparkJars))
+
   assign(
     ".sparkRjsc",
     J("edu.berkeley.cs.amplab.sparkr.RRDD",
       "createSparkContext",
       master,
       appName,
       as.character(sparkHome),
-      .jarray(as.character(.sparkREnv$assemblyJarPath),
-               "java/lang/String"),
-      hm),
+      .jarray(jars, "java/lang/String"),
+      sparkEnvirMap,
+      sparkExecutorEnvMap),
     envir=.sparkREnv
   )
 

pkg/src/src/main/scala/edu/berkeley/cs/amplab/sparkr/RRDD.scala

@@ -181,15 +181,19 @@ object RRDD {
       appName: String,
       sparkHome: String,
       jars: Array[String],
-      vars: JMap[Object, Object]): JavaSparkContext = {
+      sparkEnvirMap: JMap[Object, Object],
+      sparkExecutorEnvMap: JMap[Object, Object]): JavaSparkContext = {
 
     val sparkConf = new SparkConf().setMaster(master)
                                    .setAppName(appName)
                                    .setSparkHome(sparkHome)
                                    .setJars(jars)
-    for ( (name, value) <- vars) {
+    for ((name, value) <- sparkEnvirMap) {
       sparkConf.set(name.asInstanceOf[String], value.asInstanceOf[String])
     }
+    for ((name, value) <- sparkExecutorEnvMap) {
+      sparkConf.setExecutorEnv(name.asInstanceOf[String], value.asInstanceOf[String])
+    }
     new JavaSparkContext(sparkConf)
   }