Pass submit parameters through system properties

This decouples SparkSubmitArgument from StandaloneRestClient such
that the latter can now be launched as a main class just like any
other invocations of Spark submit.

Author: Andrew Or

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

@@ -49,7 +49,7 @@ private[spark] class ClientArguments(args: Array[String]) {
 
   parse(args.toList)
 
-  def parse(args: List[String]): Unit = args match {
+  private def parse(args: List[String]): Unit = args match {
     case ("--cores" | "-c") :: IntParam(value) :: tail =>
       cores = value
       parse(tail)

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

@@ -138,28 +138,22 @@ object SparkSubmit {
    */
   private[spark] def submit(args: SparkSubmitArguments): Unit = {
     val (childArgs, childClasspath, sysProps, childMainClass) = prepareSubmitEnvironment(args)
-    /*
-     * In standalone cluster mode, there are two submission gateways:
-     *   (1) The traditional Akka gateway using o.a.s.deploy.Client as a wrapper
-     *   (2) The new REST-based gateway introduced in Spark 1.3
-     * The latter is the default behavior as of Spark 1.3, but Spark submit will fail over
-     * to use the legacy gateway if the master endpoint turns out to be not a REST server.
-     */
-    if (args.isStandaloneCluster) {
+     // In standalone cluster mode, there are two submission gateways:
+     //   (1) The traditional Akka gateway using o.a.s.deploy.Client as a wrapper
+     //   (2) The new REST-based gateway introduced in Spark 1.3
+     // The latter is the default behavior as of Spark 1.3, but Spark submit will fail over
+     // to use the legacy gateway if the master endpoint turns out to be not a REST server.
+    if (args.isStandaloneCluster && args.useRest) {
       try {
         printStream.println("Running Spark using the REST application submission protocol.")
-        val client = new StandaloneRestClient
-        val response = client.createSubmission(args)
-        response match {
-          case s: CreateSubmissionResponse => handleRestResponse(s)
-          case r => handleUnexpectedRestResponse(r)
-        }
+        runMain(childArgs, childClasspath, sysProps, childMainClass)
       } catch {
         // Fail over to use the legacy submission gateway
         case e: SubmitRestConnectionException =>
           printStream.println(s"Master endpoint ${args.master} was not a " +
             s"REST server. Falling back to legacy submission gateway instead.")
-          runMain(childArgs, childClasspath, sysProps, childMainClass)
+          args.useRest = false
+          submit(args)
       }
     // In all other modes, just run the main class as prepared
     } else {
@@ -172,9 +166,8 @@ object SparkSubmit {
    * This returns a 4-tuple:
    *   (1) the arguments for the child process,
    *   (2) a list of classpath entries for the child,
-   *   (3) a list of system properties and env vars, and
+   *   (3) a map of system properties, and
    *   (4) the main class for the child
-   * In standalone cluster mode, this mutates the original arguments passed in.
    * Exposed for testing.
    */
   private[spark] def prepareSubmitEnvironment(args: SparkSubmitArguments)
@@ -319,8 +312,12 @@ object SparkSubmit {
       // Standalone cluster only
       OptionAssigner(args.jars, STANDALONE, CLUSTER, sysProp = "spark.jars"),
       OptionAssigner(args.ivyRepoPath, STANDALONE, CLUSTER, sysProp = "spark.jars.ivy"),
-      OptionAssigner(args.driverMemory, STANDALONE, CLUSTER, clOption = "--memory"),
-      OptionAssigner(args.driverCores, STANDALONE, CLUSTER, clOption = "--cores"),
+      OptionAssigner(args.driverMemory, STANDALONE, CLUSTER, sysProp = "spark.driver.memory"),
+      OptionAssigner(args.driverCores, STANDALONE, CLUSTER, sysProp = "spark.driver.cores"),
+      OptionAssigner(args.primaryResource, STANDALONE, CLUSTER, sysProp = "spark.app.resource"),
+      OptionAssigner(args.mainClass, STANDALONE, CLUSTER, sysProp = "spark.app.mainClass"),
+      OptionAssigner(args.supervise.toString, STANDALONE, CLUSTER,
+        sysProp = "spark.driver.supervise"),
 
       // Yarn client only
       OptionAssigner(args.queue, YARN, CLIENT, sysProp = "spark.yarn.queue"),
@@ -382,15 +379,22 @@ object SparkSubmit {
       sysProps.put("spark.jars", jars.mkString(","))
     }
 
-    // In standalone-cluster mode, use Client as a wrapper around the user class
-    // Note that we won't actually launch this class if we're using the REST protocol
+    // In standalone cluster mode, use the REST client to submit the application (Spark 1.3+).
+    // All parameters except application arguments are expected to be passed to the main class
+    // through system properties.
     if (args.isStandaloneCluster) {
-      childMainClass = "org.apache.spark.deploy.Client"
-      if (args.supervise) {
-        childArgs += "--supervise"
+      if (args.useRest) {
+        childMainClass = "org.apache.spark.deploy.rest.StandaloneRestClient"
+      } else {
+        // In legacy standalone cluster mode, use Client as a wrapper around the user class
+        childMainClass = "org.apache.spark.deploy.Client"
+        if (args.supervise) { childArgs += "--supervise" }
+        Option(args.driverMemory).foreach { m => childArgs += ("--memory", m) }
+        Option(args.driverCores).foreach { c => childArgs += ("--cores", c) }
+        childArgs += "launch"
+        childArgs += (args.master, args.primaryResource, args.mainClass)
       }
-      childArgs += "launch"
-      childArgs += (args.master, args.primaryResource, args.mainClass)
+      // Whether or not we use REST, pass application arguments through the command line
       if (args.childArgs != null) {
         childArgs ++= args.childArgs
       }
@@ -453,16 +457,6 @@ object SparkSubmit {
       sysProps("spark.submit.pyFiles") = formattedPyFiles
     }
 
-    // NOTE: If we are using the REST gateway, we will use the original arguments directly.
-    // Since we mutate the values of some configs in this method, we must update the
-    // corresponding fields in the original SparkSubmitArguments to reflect these changes.
-    if (args.isStandaloneCluster) {
-      args.sparkProperties.clear()
-      args.sparkProperties ++= sysProps
-      sysProps.get("spark.jars").foreach { args.jars = _ }
-      sysProps.get("spark.files").foreach { args.files = _ }
-    }
-
     (childArgs, childClasspath, sysProps, childMainClass)
   }
 

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

@@ -59,6 +59,7 @@ private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, St
   val sparkProperties: HashMap[String, String] = new HashMap[String, String]()
 
   // Standalone cluster mode only
+  var useRest: Boolean = true
   var supervise: Boolean = false
   var driverCores: String = null
   var driverToKill: String = null

core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestClient.scala

@@ -24,8 +24,7 @@ import scala.io.Source
 
 import com.google.common.base.Charsets
 
-import org.apache.spark.{Logging, SPARK_VERSION => sparkVersion}
-import org.apache.spark.deploy.SparkSubmitArguments
+import org.apache.spark.{Logging, SparkConf, SPARK_VERSION => sparkVersion}
 
 /**
  * A client that submits applications to the standalone Master using a REST protocol.
@@ -55,12 +54,15 @@ private[spark] class StandaloneRestClient extends Logging {
    * If the submission was successful, poll the status of the submission and report
    * it to the user. Otherwise, report the error message provided by the server.
    */
-  def createSubmission(args: SparkSubmitArguments): SubmitRestProtocolResponse = {
-    logInfo(s"Submitting a request to launch a driver in ${args.master}.")
-    validateSubmitArgs(args)
-    val master = args.master
+  def createSubmission(
+      master: String,
+      appArgs: Array[String],
+      sparkProperties: Map[String, String],
+      environmentVariables: Map[String, String]): SubmitRestProtocolResponse = {
+    logInfo(s"Submitting a request to launch a driver in $master.")
+    validateMaster(master)
     val url = getSubmitUrl(master)
-    val request = constructSubmitRequest(args)
+    val request = constructSubmitRequest(appArgs, sparkProperties, environmentVariables)
     val response = postJson(url, request.toJson)
     response match {
       case s: CreateSubmissionResponse => reportSubmissionStatus(master, s)
@@ -182,36 +184,16 @@ private[spark] class StandaloneRestClient extends Logging {
     }
   }
 
-  /** Throw an exception if this is not standalone cluster mode. */
-  private def validateSubmitArgs(args: SparkSubmitArguments): Unit = {
-    if (!args.isStandaloneCluster) {
-      throw new IllegalArgumentException(
-        "This REST client is only supported in standalone cluster mode.")
-    }
-  }
-
   /** Construct a message that captures the specified parameters for submitting an application. */
-  private def constructSubmitRequest(args: SparkSubmitArguments): CreateSubmissionRequest = {
+  def constructSubmitRequest(
+      appArgs: Array[String],
+      sparkProperties: Map[String, String],
+      environmentVariables: Map[String, String]): CreateSubmissionRequest = {
     val message = new CreateSubmissionRequest
     message.clientSparkVersion = sparkVersion
-    message.appName = args.name
-    message.appResource = args.primaryResource
-    message.mainClass = args.mainClass
-    message.jars = args.jars
-    message.files = args.files
-    message.driverMemory = args.driverMemory
-    message.driverCores = args.driverCores
-    message.driverExtraJavaOptions = args.driverExtraJavaOptions
-    message.driverExtraClassPath = args.driverExtraClassPath
-    message.driverExtraLibraryPath = args.driverExtraLibraryPath
-    message.superviseDriver = args.supervise.toString
-    message.executorMemory = args.executorMemory
-    message.totalExecutorCores = args.totalExecutorCores
-    args.childArgs.foreach(message.addAppArg)
-    args.sparkProperties.foreach { case (k, v) => message.setSparkProperty(k, v) }
-    sys.env.foreach { case (k, v) =>
-      if (k.startsWith("SPARK_")) { message.setEnvironmentVariable(k, v) }
-    }
+    message.appArgs = appArgs
+    message.sparkProperties = sparkProperties
+    message.environmentVariables = environmentVariables
     message.validate()
     message
   }
@@ -273,8 +255,23 @@ private[spark] class StandaloneRestClient extends Logging {
   }
 }
 
-private object StandaloneRestClient {
+private[spark] object StandaloneRestClient {
   val REPORT_DRIVER_STATUS_INTERVAL = 1000
   val REPORT_DRIVER_STATUS_MAX_TRIES = 10
   val PROTOCOL_VERSION = "v1"
+
+  /**
+   * Submit an application, assuming parameters are specified through system properties.
+   * Usage: StandaloneRestClient [app args*]
+   */
+  def main(args: Array[String]): Unit = {
+    val client = new StandaloneRestClient
+    val appArgs = args.slice(1, args.size)
+    val master = sys.props.get("spark.master").getOrElse {
+      throw new IllegalArgumentException("'spark.master' must be set.")
+    }
+    val sparkProperties = new SparkConf().getAll.toMap
+    val environmentVariables = sys.env.filter { case (k, _) => k.startsWith("SPARK_") }
+    client.createSubmission(master, appArgs, sparkProperties, environmentVariables)
+  }
 }

core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala

@@ -146,7 +146,10 @@ private[spark] abstract class StandaloneRestServlet extends HttpServlet with Log
     val clientSideJson = parse(requestJson)
     val serverSideJson = parse(requestMessage.toJson)
     val Diff(_, _, unknown) = clientSideJson.diff(serverSideJson)
-    unknown.asInstanceOf[JObject].obj.map { case (k, _) => k }.toArray
+    unknown match {
+      case j: JObject => j.obj.map { case (k, _) => k }.toArray
+      case _ => Array.empty[String] // No difference
+    }
   }
 
   /** Return a human readable String representation of the exception. */
@@ -334,43 +337,30 @@ private[spark] class SubmitRequestServlet(master: Master) extends StandaloneRest
    * cluster mode yet.
    */
   private def buildDriverDescription(request: CreateSubmissionRequest): DriverDescription = {
+    val sparkProperties = request.sparkProperties
+
     // Required fields, including the main class because python is not yet supported
-    val appName = request.appName
-    val appResource = request.appResource
-    val mainClass = request.mainClass
-    if (mainClass == null) {
-      throw new SubmitRestMissingFieldException("Main class must be set in submit request.")
+    val appResource = sparkProperties.get("spark.app.resource").getOrElse {
+      throw new SubmitRestMissingFieldException("Main application resource is missing.")
+    }
+    val mainClass = sparkProperties.get("spark.app.mainClass").getOrElse {
+      throw new SubmitRestMissingFieldException("Main class is missing.")
     }
 
     // Optional fields
-    val jars = Option(request.jars)
-    val files = Option(request.files)
-    val driverMemory = Option(request.driverMemory)
-    val driverCores = Option(request.driverCores)
-    val driverExtraJavaOptions = Option(request.driverExtraJavaOptions)
-    val driverExtraClassPath = Option(request.driverExtraClassPath)
-    val driverExtraLibraryPath = Option(request.driverExtraLibraryPath)
-    val superviseDriver = Option(request.superviseDriver)
-    val executorMemory = Option(request.executorMemory)
-    val totalExecutorCores = Option(request.totalExecutorCores)
+    val driverMemory = sparkProperties.get("spark.driver.memory")
+    val driverCores = sparkProperties.get("spark.driver.cores")
+    val driverExtraJavaOptions = sparkProperties.get("spark.driver.extraJavaOptions")
+    val driverExtraClassPath = sparkProperties.get("spark.driver.extraClassPath")
+    val driverExtraLibraryPath = sparkProperties.get("spark.driver.extraLibraryPath")
+    val superviseDriver = sparkProperties.get("spark.driver.supervise")
     val appArgs = request.appArgs
-    val sparkProperties = request.sparkProperties
     val environmentVariables = request.environmentVariables
 
-    // Translate all fields to the relevant Spark properties
+    // Construct driver description and submit it
     val conf = new SparkConf(false)
       .setAll(sparkProperties)
       .set("spark.master", master.masterUrl)
-      .set("spark.app.name", appName)
-    jars.foreach { j => conf.set("spark.jars", j) }
-    files.foreach { f => conf.set("spark.files", f) }
-    driverExtraJavaOptions.foreach { j => conf.set("spark.driver.extraJavaOptions", j) }
-    driverExtraClassPath.foreach { cp => conf.set("spark.driver.extraClassPath", cp) }
-    driverExtraLibraryPath.foreach { lp => conf.set("spark.driver.extraLibraryPath", lp) }
-    executorMemory.foreach { m => conf.set("spark.executor.memory", m) }
-    totalExecutorCores.foreach { c => conf.set("spark.cores.max", c) }
-
-    // Construct driver description and submit it
     val extraClassPath = driverExtraClassPath.toSeq.flatMap(_.split(File.pathSeparator))
     val extraLibraryPath = driverExtraLibraryPath.toSeq.flatMap(_.split(File.pathSeparator))
     val extraJavaOpts = driverExtraJavaOptions.map(Utils.splitCommandString).getOrElse(Seq.empty)

core/src/main/scala/org/apache/spark/deploy/rest/SubmitRestProtocolMessage.scala

@@ -83,8 +83,7 @@ private[spark] abstract class SubmitRestProtocolMessage {
   /** Assert that the specified field is set in this message. */
   protected def assertFieldIsSet(value: String, name: String): Unit = {
     if (value == null) {
-      throw new SubmitRestMissingFieldException(
-        s"Field '$name' is missing in message $messageType.")
+      throw new SubmitRestMissingFieldException(s"'$name' is missing in message $messageType.")
     }
   }
 
@@ -93,7 +92,7 @@ private[spark] abstract class SubmitRestProtocolMessage {
     if (value != null) {
       Try(value.toBoolean).getOrElse {
         throw new SubmitRestProtocolException(
-          s"Field '$name' expected boolean value: actual was '$value'.")
+          s"'$name' expected boolean value: actual was '$value'.")
       }
     }
   }
@@ -103,7 +102,7 @@ private[spark] abstract class SubmitRestProtocolMessage {
     if (value != null) {
       Try(value.toInt).getOrElse {
         throw new SubmitRestProtocolException(
-          s"Field '$name' expected numeric value: actual was '$value'.")
+          s"'$name' expected numeric value: actual was '$value'.")
       }
     }
   }
@@ -116,7 +115,7 @@ private[spark] abstract class SubmitRestProtocolMessage {
     if (value != null) {
       Try(Utils.memoryStringToMb(value)).getOrElse {
         throw new SubmitRestProtocolException(
-          s"Field '$name' expected memory value: actual was '$value'.")
+          s"'$name' expected memory value: actual was '$value'.")
       }
     }
   }