Use spark-submit script in SparkLauncher.

To reduce the surface for backwards compatibility, make SparkLauncher
launch child processes through the spark-submit script. This way, the
library doesn't need to know about the distribution layout of the Spark
version being launched, as long as it has a "bin/spark-submit" script.

Do some cleanup and fix a couple of issues that the change unearthed.

Author: Marcelo Vanzin

launcher/src/main/java/org/apache/spark/launcher/Main.java

@@ -88,16 +88,11 @@ public static void main(String[] argsArray) throws Exception {
   /**
    * Prepare a command line for execution from a Windows batch script.
    *
-   * Two things need to be done:
+   * The method quotes all arguments so that spaces are handled as expected. Quotes within arguments
+   * are "double quoted" (which is batch for escaping a quote). This page has more details about
+   * quoting and other batch script fun stuff: http://ss64.com/nt/syntax-esc.html
    *
-   * - If a custom library path is needed, extend PATH to add it. Based on:
-   *   http://superuser.com/questions/223104/setting-environment-variable-for-just-one-command-in-windows-cmd-exe
-   *
-   * - Quote all arguments so that spaces are handled as expected. Quotes within arguments are
-   *   "double quoted" (which is batch for escaping a quote). This page has more details about
-   *   quoting and other batch script fun stuff: http://ss64.com/nt/syntax-esc.html
-   *
-   * The command is executed using "cmd /c" and formatted as single line, since that's the
+   * The command is executed using "cmd /c" and formatted in single line, since that's the
    * easiest way to consume this from a batch script (see spark-class2.cmd).
    */
   private static String prepareForWindows(List<String> cmd, Map<String, String> childEnv) {

launcher/src/main/java/org/apache/spark/launcher/SparkLauncher.java

@@ -282,10 +282,12 @@ public SparkLauncher setVerbose(boolean verbose) {
    * @return A process handle for the Spark app.
    */
   public Process launch() throws IOException {
-    Map<String, String> procEnv = new HashMap<String, String>(childEnv);
-    List<String> cmd = buildSparkSubmitCommand(procEnv);
+    List<String> cmd = new ArrayList<String>();
+    cmd.add(join(File.separator, getSparkHome(), "bin", "spark-submit"));
+    cmd.addAll(buildSparkSubmitArgs());
+
     ProcessBuilder pb = new ProcessBuilder(cmd.toArray(new String[cmd.size()]));
-    for (Map.Entry<String, String> e : procEnv.entrySet()) {
+    for (Map.Entry<String, String> e : childEnv.entrySet()) {
       pb.environment().put(e.getKey(), e.getValue());
     }
     return pb.start();
@@ -406,7 +408,7 @@ List<String> buildClassPath(String appClassPath) throws IOException {
       assemblyJar = new JarFile(assembly);
       needsDataNucleus = assemblyJar.getEntry("org/apache/hadoop/hive/ql/exec/") != null;
     } catch (IOException ioe) {
-      if (ioe.getMessage().indexOf("invalid CEN header") > 0) {
+      if (ioe.getMessage().indexOf("invalid CEN header") >= 0) {
         System.err.println(
           "Loading Spark jar failed.\n" +
           "This is likely because Spark was compiled with Java 7 and run\n" +
@@ -565,40 +567,6 @@ List<String> buildSparkSubmitArgs() {
     return args;
   }
 
-  List<String> buildSparkSubmitCommand(Map<String, String> env) throws IOException {
-    // Load the properties file and check whether spark-submit will be running the app's driver
-    // or just launching a cluster app. When running the driver, the JVM's argument will be
-    // modified to cover the driver's configuration.
-    Properties props = loadPropertiesFile();
-    boolean isClientMode = isClientMode(props);
-    String extraClassPath = isClientMode ? find(DRIVER_EXTRA_CLASSPATH, conf, props) : null;
-
-    List<String> cmd = buildJavaCommand(extraClassPath);
-    addOptionString(cmd, System.getenv("SPARK_SUBMIT_OPTS"));
-    addOptionString(cmd, System.getenv("SPARK_JAVA_OPTS"));
-
-    if (isClientMode) {
-      // Figuring out where the memory value come from is a little tricky due to precedence.
-      // Precedence is observed in the following order:
-      // - explicit configuration (setConf()), which also covers --driver-memory cli argument.
-      // - properties file.
-      // - SPARK_DRIVER_MEMORY env variable
-      // - SPARK_MEM env variable
-      // - default value (512m)
-      String memory = firstNonEmpty(find(DRIVER_MEMORY, conf, props),
-        System.getenv("SPARK_DRIVER_MEMORY"), System.getenv("SPARK_MEM"), DEFAULT_MEM);
-      cmd.add("-Xms" + memory);
-      cmd.add("-Xmx" + memory);
-      addOptionString(cmd, find(DRIVER_EXTRA_JAVA_OPTIONS, conf, props));
-      mergeEnvPathList(env, getLibPathEnvName(), find(DRIVER_EXTRA_LIBRARY_PATH, conf, props));
-    }
-
-    addPermGenSizeOpt(cmd);
-    cmd.add("org.apache.spark.deploy.SparkSubmit");
-    cmd.addAll(buildSparkSubmitArgs());
-    return cmd;
-  }
-
   String getSparkHome() {
     String path = getenv(ENV_SPARK_HOME);
     checkState(path != null,
@@ -644,14 +612,6 @@ String getenv(String key) {
     return firstNonEmpty(childEnv.get(key), System.getenv(key));
   }
 
-  private boolean isClientMode(Properties userProps) {
-    String userMaster = firstNonEmpty(master, (String) userProps.get(SPARK_MASTER));
-    return userMaster == null ||
-      "client".equals(deployMode) ||
-      "yarn-client".equals(userMaster) ||
-      (deployMode == null && !userMaster.startsWith("yarn-"));
-  }
-
   private String findAssembly(String scalaVersion) {
     String sparkHome = getSparkHome();
     File libdir;

launcher/src/main/java/org/apache/spark/launcher/SparkSubmitCommandBuilder.java

@@ -21,12 +21,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 import static org.apache.spark.launcher.CommandBuilderUtils.*;
 
@@ -89,10 +86,44 @@ public List<String> buildCommand(Map<String, String> env) throws IOException {
     if (PYSPARK_SHELL.equals(appResource)) {
       return buildPySparkShellCommand(env);
     } else {
-      return super.buildSparkSubmitCommand(env);
+      return buildSparkSubmitCommand(env);
     }
   }
 
+  private List<String> buildSparkSubmitCommand(Map<String, String> env) throws IOException {
+    // Load the properties file and check whether spark-submit will be running the app's driver
+    // or just launching a cluster app. When running the driver, the JVM's argument will be
+    // modified to cover the driver's configuration.
+    Properties props = loadPropertiesFile();
+    boolean isClientMode = isClientMode(props);
+    String extraClassPath = isClientMode ? find(DRIVER_EXTRA_CLASSPATH, conf, props) : null;
+
+    List<String> cmd = buildJavaCommand(extraClassPath);
+    addOptionString(cmd, System.getenv("SPARK_SUBMIT_OPTS"));
+    addOptionString(cmd, System.getenv("SPARK_JAVA_OPTS"));
+
+    if (isClientMode) {
+      // Figuring out where the memory value come from is a little tricky due to precedence.
+      // Precedence is observed in the following order:
+      // - explicit configuration (setConf()), which also covers --driver-memory cli argument.
+      // - properties file.
+      // - SPARK_DRIVER_MEMORY env variable
+      // - SPARK_MEM env variable
+      // - default value (512m)
+      String memory = firstNonEmpty(find(DRIVER_MEMORY, conf, props),
+        System.getenv("SPARK_DRIVER_MEMORY"), System.getenv("SPARK_MEM"), DEFAULT_MEM);
+      cmd.add("-Xms" + memory);
+      cmd.add("-Xmx" + memory);
+      addOptionString(cmd, find(DRIVER_EXTRA_JAVA_OPTIONS, conf, props));
+      mergeEnvPathList(env, getLibPathEnvName(), find(DRIVER_EXTRA_LIBRARY_PATH, conf, props));
+    }
+
+    addPermGenSizeOpt(cmd);
+    cmd.add("org.apache.spark.deploy.SparkSubmit");
+    cmd.addAll(buildSparkSubmitArgs());
+    return cmd;
+  }
+
   private List<String> buildPySparkShellCommand(Map<String, String> env) throws IOException {
     // For backwards compatibility, if a script is specified in
     // the pyspark command line, then run it using spark-submit.
@@ -139,6 +170,14 @@ private List<String> buildPySparkShellCommand(Map<String, String> env) throws IO
     return pyargs;
   }
 
+  private boolean isClientMode(Properties userProps) {
+    String userMaster = firstNonEmpty(master, (String) userProps.get(SPARK_MASTER));
+    return userMaster == null ||
+      "client".equals(deployMode) ||
+      "yarn-client".equals(userMaster) ||
+      (deployMode == null && !userMaster.startsWith("yarn-"));
+  }
+
   /**
    * Quotes a string so that it can be used in a command string and be parsed back into a single
    * argument by python's "shlex.split()" function.
@@ -172,19 +211,25 @@ protected boolean handle(String opt, String value) {
         driverArgs.add(opt);
         driverArgs.add(value);
       } else if (opt.equals(DRIVER_MEMORY)) {
-        setConf(DRIVER_MEMORY, value);
+        setConf(SparkLauncher.DRIVER_MEMORY, value);
         driverArgs.add(opt);
         driverArgs.add(value);
       } else if (opt.equals(DRIVER_JAVA_OPTIONS)) {
-        setConf(DRIVER_EXTRA_JAVA_OPTIONS, value);
+        setConf(SparkLauncher.DRIVER_EXTRA_JAVA_OPTIONS, value);
         driverArgs.add(opt);
         driverArgs.add(value);
       } else if (opt.equals(DRIVER_LIBRARY_PATH)) {
-        setConf(DRIVER_EXTRA_LIBRARY_PATH, value);
+        setConf(SparkLauncher.DRIVER_EXTRA_LIBRARY_PATH, value);
         driverArgs.add(opt);
         driverArgs.add(value);
       } else if (opt.equals(DRIVER_CLASS_PATH)) {
-        setConf(DRIVER_EXTRA_CLASSPATH, value);
+        setConf(SparkLauncher.DRIVER_EXTRA_CLASSPATH, value);
+        driverArgs.add(opt);
+        driverArgs.add(value);
+      } else if (opt.equals(CONF)) {
+        String[] conf = value.split("=", 2);
+        checkArgument(conf.length == 2, "Invalid argument to %s: %s", CONF, value);
+        handleConf(conf[0], conf[1]);
         driverArgs.add(opt);
         driverArgs.add(value);
       } else if (opt.equals(CLASS)) {
@@ -227,6 +272,18 @@ protected void handleExtraArgs(List<String> extra) {
       }
     }
 
+    private void handleConf(String key, String value) {
+      List<String> driverJvmKeys = Arrays.asList(
+        SparkLauncher.DRIVER_EXTRA_CLASSPATH,
+        SparkLauncher.DRIVER_EXTRA_JAVA_OPTIONS,
+        SparkLauncher.DRIVER_EXTRA_LIBRARY_PATH,
+        SparkLauncher.DRIVER_MEMORY);
+
+      if (driverJvmKeys.contains(key)) {
+        setConf(key, value);
+      }
+    }
+
   }
 
 }