Add support for SparkR shell to use spark-submit

This ensures that SparkConf options are read in both
in batch and interactive modes

Author: shivaram

pkg/R/sparkR.R

@@ -74,7 +74,7 @@ sparkR.stop <- function(env) {
 #'}
 
 sparkR.init <- function(
-  master = "local",
+  master = "",
   appName = "SparkR",
   sparkHome = Sys.getenv("SPARK_HOME"),
   sparkEnvir = list(),
@@ -101,10 +101,21 @@ sparkR.init <- function(
   if (sparkRExistingPort != "") {
     sparkRBackendPort <- sparkRExistingPort
   } else {
-    launchBackend(classPath = cp,
-                  mainClass = "edu.berkeley.cs.amplab.sparkr.SparkRBackend",
-                  args = as.character(sparkRBackendPort),
-                  javaOpts = paste("-Xmx", sparkMem, sep = ""))
+    if (Sys.getenv("SPARKR_USE_SPARK_SUBMIT", "") == "") {
+      launchBackend(classPath = cp,
+                    mainClass = "edu.berkeley.cs.amplab.sparkr.SparkRBackend",
+                    args = as.character(sparkRBackendPort),
+                    javaOpts = paste("-Xmx", sparkMem, sep = ""))
+    } else {
+      # TODO: We should deprecate sparkJars and ask users to add it to the
+      # command line (using --jars) which is picked up by SparkSubmit
+      launchBackendSparkSubmit(
+          mainClass = "edu.berkeley.cs.amplab.sparkr.SparkRBackend",
+          args = as.character(sparkRBackendPort),
+          appJar = .sparkREnv$assemblyJarPath,
+          sparkHome = sparkHome,
+          sparkSubmitOpts = Sys.getenv("SPARKR_SUBMIT_ARGS", ""))
+    }
     Sys.sleep(2) # Wait for backend to come up
   }
   .sparkREnv$sparkRBackendPort <- sparkRBackendPort

pkg/R/sparkRClient.R

@@ -37,3 +37,30 @@ launchBackend <- function(
   cat("Launching java with command ", java_bin, " ", combinedArgs, "\n")
   invisible(system2(java_bin, combinedArgs, wait = F))
 }
+
+launchBackendSparkSubmit <- function(
+    mainClass,
+    args,
+    appJar,
+    sparkHome,
+    sparkSubmitOpts) {
+  if (.Platform$OS.type == "unix") {
+    spark_submit_bin_name = "spark-submit"
+  } else {
+    spark_submit_bin_name = "spark-submit.cmd"
+  }
+
+  if (sparkHome != "") {
+    spark_submit_bin <- file.path(sparkHome, "bin", spark_submit_bin_name)
+  } else {
+    spark_submit_bin <- spark_submit_bin_name
+  }
+
+  # Since this function is only used while launching R shell using spark-submit,
+  # the format we need to construct is
+  # spark-submit --class <mainClass> <sparkSubmitOpts> <jarFile> <appOpts>
+
+  combinedArgs <- paste("--class", mainClass, sparkSubmitOpts, appJar, args, sep = " ")
+  cat("Launching java with spark-submit command ", spark_submit_bin, " ", combinedArgs, "\n")
+  invisible(system2(spark_submit_bin, combinedArgs, wait = F))
+}

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

@@ -357,10 +357,19 @@ object RRDD {
       sparkEnvirMap: JMap[Object, Object],
       sparkExecutorEnvMap: JMap[Object, Object]): JavaSparkContext = {
 
-    val sparkConf = new SparkConf().setMaster(master)
-                                   .setAppName(appName)
+    val sparkConf = new SparkConf().setAppName(appName)
                                    .setSparkHome(sparkHome)
                                    .setJars(jars)
+
+    // Override `master` if we have a user-specified value
+    if (master != "") {
+      sparkConf.setMaster(master)
+    } else {
+      // If conf has no master set it to "local" to maintain
+      // backwards compatibility
+      sparkConf.setIfMissing("spark.master", "local")
+    }
+
     for ((name, value) <- sparkEnvirMap) {
       sparkConf.set(name.asInstanceOf[String], value.asInstanceOf[String])
     }

sparkR

@@ -28,7 +28,7 @@ cat > /tmp/sparkR.profile << EOF
   Sys.setenv(NOAWT=1)
   .libPaths(c(paste(projecHome,"/lib", sep=""), .libPaths()))
   require(SparkR)
-  sc <- sparkR.init(Sys.getenv("MASTER", unset = "local"))
+  sc <- sparkR.init(Sys.getenv("MASTER", unset = ""))
   assign("sc", sc, envir=.GlobalEnv)
   cat("\n Welcome to SparkR!")
   cat("\n Spark context is available as sc\n")

sparkR-submit

@@ -33,6 +33,26 @@ fi
 
 export R_PROFILE_USER="/tmp/sparkR.profile"
 
+# Build up arguments list manually to preserve quotes and backslashes.
+SUBMIT_USAGE_FUNCTION=usage
+gatherSparkSubmitOpts "$@"
+
+SPARKR_SUBMIT_ARGS=""
+whitespace="[[:space:]]"
+for i in "${SUBMISSION_OPTS[@]}"
+do
+  if [[ $i =~ \" ]]; then i=$(echo $i | sed 's/\"/\\\"/g'); fi
+  if [[ $i =~ $whitespace ]]; then i=\"$i\"; fi
+  SPARKR_SUBMIT_ARGS="$SPARKR_SUBMIT_ARGS $i"
+done
+export SPARKR_SUBMIT_ARGS
+export SPARKR_USE_SPARK_SUBMIT=1
+
+NUM_APPLICATION_OPTS=${#APPLICATION_OPTS[@]}
+
+# If a R file is provided, directly run spark-submit. 
+if [[ $NUM_APPLICATION_OPTS -gt 0 && "${APPLICATION_OPTS[0]}" =~ \.R$ ]]; then
+
 cat > /tmp/sparkR.profile << EOF
   .First <- function() {
   projecHome <- Sys.getenv("PROJECT_HOME")
@@ -41,16 +61,30 @@ cat > /tmp/sparkR.profile << EOF
 }
 EOF
 
-# Build up arguments list manually to preserve quotes and backslashes.
-SUBMIT_USAGE_FUNCTION=usage
-gatherSparkSubmitOpts "$@"
-
-# If a R file is provided, directly run spark-submit. 
-if [[ "${APPLICATION_OPTS[0]}" =~ \.R$ ]]; then
   primary="${APPLICATION_OPTS[0]}"
   shift
   # Set the main class to SparkRRunner and add the primary R file to --files to make sure its copied to the cluster
+  echo "Running $SPARK_HOME/bin/spark-submit --class edu.berkeley.cs.amplab.sparkr.SparkRRunner --files $primary ${SUBMISSION_OPTS[@]} $SPARKR_JAR_FILE $primary" "${APPLICATION_OPTS[@]:1}"
   exec "$SPARK_HOME"/bin/spark-submit --class edu.berkeley.cs.amplab.sparkr.SparkRRunner --files "$primary" "${SUBMISSION_OPTS[@]}" "$SPARKR_JAR_FILE" "$primary" "${APPLICATION_OPTS[@]:1}"
 else
-  echo "sparkR-submit can only be used to run R programs. Please use sparkR to launch a shell."
+
+  # If we don't have an R file to run, initialize context and run R
+cat > /tmp/sparkR.profile << EOF
+.First <- function() {
+  projecHome <- Sys.getenv("PROJECT_HOME")
+  Sys.setenv(NOAWT=1)
+  .libPaths(c(paste(projecHome,"/lib", sep=""), .libPaths()))
+  require(SparkR)
+  sc <- sparkR.init(Sys.getenv("MASTER", unset = ""))
+  assign("sc", sc, envir=.GlobalEnv)
+  cat("\n Welcome to SparkR!")
+  cat("\n Spark context is available as sc\n")
+}
+EOF
+  
+  # Add SPARKR_JAR, main class etc. to SPARKR_SUBMIT_ARGS 
+  export SPARKR_SUBMIT_ARGS
+
+  R
+
 fi