Responses to review feedback.

Author: pwendell

conf/spark-env.sh.template

@@ -1,10 +1,10 @@
 #!/usr/bin/env bash
 
-# This file is sourced when running various Spark classes. 
+# This file is sourced when running various Spark programs.
 # Copy it as spark-env.sh and edit that to configure Spark for your site.
 
 # Options read when launching programs locally with 
-# ./bin/spark-example or ./bin/spark-submit
+# ./bin/run-example or ./bin/spark-submit
 # - SPARK_LOCAL_IP, to set the IP address Spark binds to on this node
 # - SPARK_PUBLIC_DNS, to set the public dns name of the driver program
 # - SPARK_CLASSPATH, default classpath entries to append
@@ -13,15 +13,15 @@
 # - SPARK_LOCAL_IP, to set the IP address Spark binds to on this node
 # - SPARK_PUBLIC_DNS, to set the public DNS name of the driver program
 # - SPARK_CLASSPATH, default classpath entries to append
-# - SPARK_LOCAL_DIRS, shuffle directories to use on this node
+# - SPARK_LOCAL_DIRS, storage directories to use on this node for shuffle and RDD data
 # - MESOS_NATIVE_LIBRARY, to point to your libmesos.so if you use Mesos
 
 # Options read in YARN client mode
 # - SPARK_YARN_APP_JAR, Path to your application’s JAR file (required)
-# - SPARK_WORKER_INSTANCES, Number of workers to start (Default: 2)
-# - SPARK_WORKER_CORES, Number of cores for the workers (Default: 1).
-# - SPARK_WORKER_MEMORY, Memory per Worker (e.g. 1000M, 2G) (Default: 1G)
-# - SPARK_MASTER_MEMORY, Memory for Master (e.g. 1000M, 2G) (Default: 512 Mb)
+# - SPARK_EXECUTOR_INSTANCES, Number of workers to start (Default: 2)
+# - SPARK_EXECUTOR_CORES, Number of cores for the workers (Default: 1).
+# - SPARK_EXECUTOR_MEMORY, Memory per Worker (e.g. 1000M, 2G) (Default: 1G)
+# - SPARK_DRIVER_MEMORY, Memory for Master (e.g. 1000M, 2G) (Default: 512 Mb)
 # - SPARK_YARN_APP_NAME, The name of your application (Default: Spark)
 # - SPARK_YARN_QUEUE, The hadoop queue to use for allocation requests (Default: ‘default’)
 # - SPARK_YARN_DIST_FILES, Comma separated list of files to be distributed with the job.

core/src/main/scala/org/apache/spark/SparkConf.scala

@@ -224,7 +224,8 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {
         throw new Exception(msg)
       }
       if (javaOpts.contains("-Xmx") || javaOpts.contains("-Xms")) {
-        val msg = s"$executorOptsKey is not allowed to alter memory settings. Was '$javaOpts'"
+        val msg = s"$executorOptsKey is not allowed to alter memory settings (was '$javaOpts'). Please use " +
+          "spark.executor.memory."
         throw new Exception(msg)
       }
     }

core/src/main/scala/org/apache/spark/deploy/Client.scala

@@ -60,7 +60,7 @@ private class ClientActor(driverArgs: ClientArguments, conf: SparkConf) extends
         val libraryPathEntries = sys.props.get("spark.driver.libraryPath").toSeq.flatMap { cp =>
           cp.split(java.io.File.pathSeparator)
         }
-        val javaOpts = sys.props.get("spark.driver.javaOpts").toSeq
+        val javaOpts = sys.props.get("spark.driver.extraJavaOptions")
         val command = new Command(mainClass, Seq("{{WORKER_URL}}", driverArgs.mainClass) ++
           driverArgs.driverOptions, env, classPathEntries, libraryPathEntries, javaOpts)
 

core/src/main/scala/org/apache/spark/deploy/Command.scala

@@ -25,5 +25,5 @@ private[spark] case class Command(
     environment: Map[String, String],
     classPathEntries: Seq[String],
     libraryPathEntries: Seq[String],
-    javaOptions: Seq[String]) {
+    extraJavaOptions: Option[String] = None) {
 }

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.deploy
 
-import java.io.{FileInputStream, PrintStream, File}
+import java.io.{IOException, FileInputStream, PrintStream, File}
 import java.net.URL
 import java.util.Properties
 
@@ -27,6 +27,7 @@ import scala.collection.JavaConversions._
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.mutable.HashMap
 import scala.collection.mutable.Map
+import org.apache.spark.SparkException
 
 /**
  * Scala code behind the spark-submit script.  The script handles setting up the classpath with
@@ -252,9 +253,14 @@ object SparkSubmit {
   }
 
   private def getDefaultProperties(file: File): Seq[(String, String)] = {
+    require(file.exists(), s"Default properties file ${file.getName} does not exist")
     val inputStream = new FileInputStream(file)
     val properties = new Properties()
-    properties.load(inputStream)
+    try {
+      properties.load(inputStream)
+    } catch {
+      case e: IOException => throw new SparkException(s"Failed when loading Spark properties file ${file.getName}", e)
+    }
     properties.stringPropertyNames().toSeq.map(k => (k, properties(k)))
   }
 }

core/src/main/scala/org/apache/spark/deploy/client/TestClient.scala

@@ -50,7 +50,7 @@ private[spark] object TestClient {
       conf = conf, securityManager = new SecurityManager(conf))
     val desc = new ApplicationDescription(
       "TestClient", Some(1), 512, Command("spark.deploy.client.TestExecutor", Seq(), Map(), Seq(),
-        Seq(), Seq()), Some("dummy-spark-home"), "ignored")
+        Seq()), Some("dummy-spark-home"), "ignored")
     val listener = new TestListener
     val client = new AppClient(actorSystem, Array(url), desc, listener, new SparkConf)
     client.start()

core/src/main/scala/org/apache/spark/deploy/worker/CommandUtils.scala

@@ -47,6 +47,8 @@ object CommandUtils extends Logging {
    */
   def buildJavaOpts(command: Command, memory: Int, sparkHome: String): Seq[String] = {
     val memoryOpts = Seq(s"-Xms${memory}M", s"-Xmx${memory}M")
+    // Note, this will coalesce multiple options into a single command component
+    val extraOpts = command.extraJavaOptions.toSeq
     val libraryOpts =
       if (command.libraryPathEntries.size > 0) {
         val joined = command.libraryPathEntries.mkString(File.pathSeparator)
@@ -63,7 +65,7 @@ object CommandUtils extends Logging {
     val userClassPath = command.classPathEntries.mkString(File.pathSeparator)
     val classPathWithUser = classPath + File.pathSeparator + userClassPath
 
-    Seq("-cp", classPathWithUser) ++ libraryOpts ++ memoryOpts ++ command.javaOptions
+    Seq("-cp", classPathWithUser) ++ libraryOpts ++ extraOpts ++ memoryOpts
   }
 
   /** Spawn a thread that will redirect a given stream to a file */

core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala

@@ -74,14 +74,14 @@ private[spark] class DriverRunner(
 
           // Make sure user application jar is on the classpath
           // TODO: If we add ability to submit multiple jars they should also be added here
-          val classPath = driverDesc.command.classPathEntries ++ Seq(s":$localJarFilename")
+          val classPath = driverDesc.command.classPathEntries ++ Seq(s"$localJarFilename")
           val newCommand = Command(
             driverDesc.command.mainClass,
             driverDesc.command.arguments.map(substituteVariables),
             driverDesc.command.environment,
             classPath,
             driverDesc.command.libraryPathEntries,
-            driverDesc.command.javaOptions)
+            driverDesc.command.extraJavaOptions)
           val command = CommandUtils.buildCommandSeq(newCommand, driverDesc.mem,
             sparkHome.getAbsolutePath)
           launchDriver(command, driverDesc.command.environment, driverDir, driverDesc.supervise)

core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala

@@ -101,7 +101,7 @@ private[spark] class ExecutorRunner(
     val command = Command(appDesc.command.mainClass,
       appDesc.command.arguments.map(substituteVariables) ++ Seq(appId), appDesc.command.environment,
       appDesc.command.classPathEntries, appDesc.command.libraryPathEntries,
-      appDesc.command.javaOptions)
+      appDesc.command.extraJavaOptions)
     CommandUtils.buildCommandSeq(command, memory, sparkHome.getAbsolutePath)
   }
 

core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala

@@ -44,14 +44,14 @@ private[spark] class SparkDeploySchedulerBackend(
     val driverUrl = "akka.tcp://spark@%s:%s/user/%s".format(
       conf.get("spark.driver.host"), conf.get("spark.driver.port"),
       CoarseGrainedSchedulerBackend.ACTOR_NAME)
-    val args = sc.conf.get("spark.executor.extraJavaOptions").split(" ") ++
-      Seq(driverUrl, "{{EXECUTOR_ID}}", "{{HOSTNAME}}",
+    val args = Seq(driverUrl, "{{EXECUTOR_ID}}", "{{HOSTNAME}}",
       "{{CORES}}", "{{WORKER_URL}}")
+    val extraJavaOpts = sc.conf.getOption("spark.executor.extraJavaOptions")
 
     // TODO (pwendell) LOOK AT THIS
     val command = Command(
       "org.apache.spark.executor.CoarseGrainedExecutorBackend", args, sc.executorEnvs,
-      Seq(), Seq(), Seq())
+      Seq(), Seq(), extraJavaOpts)
     val sparkHome = sc.getSparkHome()
     val appDesc = new ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command,
       sparkHome, sc.ui.appUIAddress, sc.eventLogger.map(_.logDir))