[SPARK-436] Remove SPARK_JAVA_OPTS from CLI (apache#129)

* [SPARK-436] remove SPARK_JAVA_OPTS from CLI

* use --conf in tests

* add spark.mesos.uris at the beginning of spark-submit, not the end

Author: mgummelt

dispatcher/cli/dcos_spark/spark_submit.py

@@ -154,10 +154,22 @@ def show_help():
     return 0
 
 
-def submit_job(dispatcher, args, docker_image, verbose=False):
-    (props, args) = partition(args.split(" "), lambda a: a.startswith("-D"))
-
-    props = props + ["-Dspark.mesos.executor.docker.image=" + docker_image]
+def submit_job(dispatcher, submit_args, docker_image, verbose=False):
+    """
+    Run spark-submit.
+
+    :param dispatcher: Spark Dispatcher URL.  Used to construct --master.
+    :type dispatcher: string
+    :param args: --submit-args value from `dcos spark run`
+    :type args: string
+    :param docker_image: Docker image to run the driver and executors in.
+    :type docker_image: string
+    :param verbose: If true, prints verbose information to stdout.
+    :type verbose: boolean
+    """
+    args = ["--conf",
+            "spark.mesos.executor.docker.image={}".format(docker_image)] + \
+        submit_args.split()
 
     hdfs_url = _get_spark_hdfs_url()
     if hdfs_url is not None:
@@ -166,10 +178,12 @@ def submit_job(dispatcher, args, docker_image, verbose=False):
             hdfs_url += '/'
         hdfs_config_url = urllib.parse.urljoin(hdfs_url, 'hdfs-site.xml')
         site_config_url = urllib.parse.urljoin(hdfs_url, 'core-site.xml')
-        props = props + ["-Dspark.mesos.uris={0},{1}".format(hdfs_config_url,
-                                                             site_config_url)]
+        args = ["--conf", "spark.mesos.uris={0},{1}".format(
+            hdfs_config_url,
+            site_config_url)] + \
+            args
 
-    response = run(dispatcher, args, verbose, props)
+    response = run(dispatcher, args, verbose)
     if response[0] is not None:
         print("Run job succeeded. Submission id: " +
               response[0]['submissionId'])
@@ -266,13 +280,16 @@ def check_java():
     return False
 
 
-def run(dispatcher, args, verbose, props=[]):
+def run(dispatcher, args, verbose):
     """
-    This method runs spark_submit with the passed in parameters.
-    ie: ./bin/spark-submit --deploy-mode cluster --class
-    org.apache.spark.examples.SparkPi --master mesos://10.127.131.174:8077
-    --executor-memory 1G --total-executor-cores 100 --driver-memory 1G
-    http://10.127.131.174:8000/spark-examples_2.10-1.3.0-SNAPSHOT.jar 30
+    Runs spark-submit.
+
+    :param dispatcher: Spark Dispatcher URL.  Used to construct --master.
+    :type dispatcher: string
+    :param args: Extra arguments to spark-submit
+    :type args: list[string]
+    :param verbose: If true, prints verbose information to stdout.
+    :type verbose: boolean
     """
     if not check_java():
         return (None, 1)
@@ -285,11 +302,12 @@ def run(dispatcher, args, verbose, props=[]):
 
     command = _get_command(dispatcher, args)
 
-    extra_env = {"SPARK_JAVA_OPTS": ' '.join(props)}
-    env = dict(os.environ, **extra_env)
-    # On Windows python 2 complains about unicode in env
-    if util.is_windows_platform() and sys.version_info[0] < 3:
-        env = dict([str(key), str(value)] for key, value in env.iteritems())
+    # On Windows, python 2 complains about unicode in env.
+    env = dict([str(key), str(value)]
+               for key, value in os.environ.iteritems()) \
+        if util.is_windows_platform() and sys.version_info[0] < 3 \
+        else os.environ
+
     process = subprocess.Popen(
         command,
         env=env,
@@ -302,9 +320,8 @@ def run(dispatcher, args, verbose, props=[]):
         proxy_thread.proxy.shutdown()
         proxy_thread.join()
 
-    if verbose is True:
+    if verbose:
         print("Ran command: " + " ".join(command))
-        print("With added env vars: {0}".format(extra_env))
         print("Stdout:")
         print(stdout)
         print("Stderr:")

docs/history-server.md

@@ -45,7 +45,7 @@ your cluster][10] and run:
 
 1.  Run jobs with the event log enabled:
 
-        $ dcos spark run --submit-args="-Dspark.eventLog.enabled=true -Dspark.eventLog.dir=hdfs://hdfs/history ... --class MySampleClass  http://external.website/mysparkapp.jar"
+        $ dcos spark run --submit-args="--conf spark.eventLog.enabled=true --conf spark.eventLog.dir=hdfs://hdfs/history ... --class MySampleClass  http://external.website/mysparkapp.jar"
 
 1.  Visit your job in the dispatcher at
 `http://<dcos_url>/service/spark/`. It will include a link to the

docs/run-job.md

@@ -43,19 +43,15 @@ you can create a configuration file.
 ## Submission
 
 All properties are submitted through the `--submit-args` option to
-`dcos spark run`. These are ultimately passed to the [`spark-submit`
-script][13].
+`dcos spark run`. These are ultimately passed to the
+[`spark-submit` script][13].  View `dcos spark run --help` for a list
+of all these options.  Example
 
 Certain common properties have their own special names. You can view
 these through `dcos spark run --help`. Here is an example of using
 `--supervise`:
 
-    $ dcos spark run --submit-args="--supervise --class MySampleClass http://external.website/mysparkapp.jar 30`
-
-Or you can set arbitrary properties as java system properties by using
-`-D<prop>=<value>`:
-
-    $ dcos spark run --submit-args="-Dspark.executor.memory=4g --class MySampleClass http://external.website/mysparkapp.jar 30`
+    $ dcos spark run --submit-args="--conf spark.executor.memory=4g --supervise --class MySampleClass http://external.website/mysparkapp.jar 30`
 
 ## Configuration file
 

docs/security.md

@@ -31,29 +31,29 @@ Follow these instructions to authenticate in strict mode:
 1. Assign Permissions
 
     First, allow Spark to run tasks as root:
-    
+
     ```
     $ curl -k -L -X PUT \
            -H "Authorization: token=$(dcos config show core.dcos_acs_token)" \
            "$(dcos config show core.dcos_url)/acs/api/v1/acls/dcos:mesos:master:task:user:root" \
            -d '{"description":"Allows root to execute tasks"}' \
            -H 'Content-Type: application/json'
-    
+
     $ curl -k -L -X PUT \
          -H "Authorization: token=$(dcos config show core.dcos_acs_token)" \
          "$(dcos config show core.dcos_url)/acs/api/v1/acls/dcos:mesos:master:task:user:root/users/${SERVICE_ACCOUNT_NAME}/create"
     ```
 
     Now, you must allow Spark to register under the desired role. This is the value used for `service.role` when installing Spark (default: `*`):
-    
+
     ```
     $ export ROLE=<service.role value>
     $ curl -k -L -X PUT \
            -H "Authorization: token=$(dcos config show core.dcos_acs_token)" \
            "$(dcos config show core.dcos_url)/acs/api/v1/acls/dcos:mesos:master:framework:role:${ROLE}" \
            -d '{"description":"Allows ${ROLE} to register as a framework with the Mesos master"}' \
            -H 'Content-Type: application/json'
-    
+
     $ curl -k -L -X PUT \
            -H "Authorization: token=$(dcos config show core.dcos_acs_token)" \
            "$(dcos config show core.dcos_url)/acs/api/v1/acls/dcos:mesos:master:framework:role:${ROLE}/users/${SERVICE_ACCOUNT_NAME}/create"
@@ -86,11 +86,11 @@ Follow these instructions to authenticate in strict mode:
 1. Submit a Job
 
     We've now installed the Spark Dispatcher, which is authenticating itself to the Mesos master. Spark jobs are also frameworks that must authenticate. The dispatcher will pass the secret along to the jobs, so all that's left to do is configure our jobs to use DC/OS authentication:
-    
+
     ```
-    $ PROPS="-Dspark.mesos.driverEnv.MESOS_MODULES=file:///opt/mesosphere/etc/mesos-scheduler-modules/dcos_authenticatee_module.json "
-    $ PROPS+="-Dspark.mesos.driverEnv.MESOS_AUTHENTICATEE=com_mesosphere_dcos_ClassicRPCAuthenticatee "
-    $ PROPS+="-Dspark.mesos.principal=<principal>"
+    $ PROPS="--conf spark.mesos.driverEnv.MESOS_MODULES=file:///opt/mesosphere/etc/mesos-scheduler-modules/dcos_authenticatee_module.json "
+    $ PROPS+="--conf spark.mesos.driverEnv.MESOS_AUTHENTICATEE=com_mesosphere_dcos_ClassicRPCAuthenticatee "
+    $ PROPS+="--conf spark.mesos.principal=<principal>"
     $ dcos spark run --submit-args="${PROPS} ..."
     ```
 

docs/troubleshooting.md

@@ -39,4 +39,4 @@ the `--verbose` flag.
 To debug authentication in a Spark job, enable Java security debug
 output:
 
-    $ dcos spark run --submit-args="-Dsun.security.krb5.debug=true..."
+    $ dcos spark run --submit-args="--conf sun.security.krb5.debug=true..."

tests/test.py

@@ -54,7 +54,7 @@ def test_jar():
     _run_tests(jar_url,
                spark_job_runner_args,
                "All tests passed",
-               {"--class": 'com.typesafe.spark.test.mesos.framework.runners.SparkJobRunner'})
+               ["--class", 'com.typesafe.spark.test.mesos.framework.runners.SparkJobRunner'])
 
 
 def test_teragen():
@@ -63,7 +63,7 @@ def test_teragen():
         _run_tests(jar_url,
                    "1g hdfs:///terasort_in",
                    "Number of records written",
-                   {"--class": "com.github.ehiggs.spark.terasort.TeraGen"})
+                   ["--class", "com.github.ehiggs.spark.terasort.TeraGen"])
 
 
 def test_python():
@@ -74,7 +74,7 @@ def test_python():
     _run_tests(python_script_url,
                "30",
                "Pi is roughly 3",
-               {"--py-files": py_file_url})
+               ["--py-files", py_file_url])
 
 
 @pytest.mark.skip(reason="must be run manually against a kerberized HDFS")
@@ -94,8 +94,10 @@ def test_kerberos():
         "http://infinity-artifacts.s3.amazonaws.com/spark/sparkjob-assembly-1.0.jar",
         "hdfs:///krb5.conf",
         "number of words in",
-        {"--class": "HDFSWordCount", "--principal":  principal, "--keytab": keytab},
-        {"sun.security.krb5.debug": "true"})
+        ["--class", "HDFSWordCount",
+         "--principal",  principal,
+         "--keytab", keytab,
+         "--conf", "sun.security.krb5.debug=true"])
 
 
 def test_r():
@@ -111,8 +113,8 @@ def test_cni():
     _run_tests(SPARK_EXAMPLES,
                "",
                "Pi is roughly 3",
-               {"--class": "org.apache.spark.examples.SparkPi"},
-               {"spark.mesos.network.name": "dcos"})
+               ["--conf", "spark.mesos.network.name=dcos",
+                "--class", "org.apache.spark.examples.SparkPi"])
 
 
 def test_s3():
@@ -123,18 +125,23 @@ def test_s3():
         s3.s3n_url('linecount.txt'),
         s3.s3n_url("linecount-out"))
 
+    args = ["--conf",
+            "spark.mesos.driverEnv.AWS_ACCESS_KEY_ID={}".format(
+                os.environ["AWS_ACCESS_KEY_ID"]),
+            "--conf",
+            "spark.mesos.driverEnv.AWS_SECRET_ACCESS_KEY={}".format(
+                os.environ["AWS_SECRET_ACCESS_KEY"]),
+            "--class", "S3Job"]
     _run_tests(_upload_file(os.environ["SCALA_TEST_JAR_PATH"]),
                app_args,
                "",
-               {"--class": "S3Job"},
-               {"spark.mesos.driverEnv.AWS_ACCESS_KEY_ID": os.environ["AWS_ACCESS_KEY_ID"],
-                "spark.mesos.driverEnv.AWS_SECRET_ACCESS_KEY": os.environ["AWS_SECRET_ACCESS_KEY"]})
+               args)
 
     assert len(list(s3.list("linecount-out"))) > 0
 
 
 def _hdfs_enabled():
-    return os.environ.get("HDFS_DISABLED") is None
+    return os.environ.get("HDFS_ENABLED") != "false"
 
 
 def _require_hdfs():
@@ -248,30 +255,28 @@ def _run_janitor(service_name):
         auth=shakedown.dcos_acs_token()))
 
 
-def _run_tests(app_url, app_args, expected_output, args={}, config={}):
-    task_id = _submit_job(app_url, app_args, args, config)
+def _run_tests(app_url, app_args, expected_output, args=[]):
+    task_id = _submit_job(app_url, app_args, args)
     LOGGER.info('Waiting for task id={} to complete'.format(task_id))
     shakedown.wait_for_task_completion(task_id)
     log = _task_log(task_id)
     LOGGER.info("task log: {}".format(log))
     assert expected_output in log
 
 
-def _submit_job(app_url, app_args, args={}, config={}):
+def _submit_job(app_url, app_args, args=[]):
     if _is_strict():
         config['spark.mesos.driverEnv.MESOS_MODULES'] = \
             'file:///opt/mesosphere/etc/mesos-scheduler-modules/dcos_authenticatee_module.json'
         config['spark.mesos.driverEnv.MESOS_AUTHENTICATEE'] = 'com_mesosphere_dcos_ClassicRPCAuthenticatee'
         config['spark.mesos.principal'] = 'service-acct'
-    args_str = ' '.join('{0} {1}'.format(k, v) for k,v in args.items())
-    config_str = ' '.join('-D{0}={1}'.format(k, v) for k,v in config.items())
-    submit_args = ' '.join(
-        arg for arg in
-        ["-Dspark.driver.memory=2g", args_str, app_url, app_args, config_str]
-        if arg != "")
+    args_str = ' '.join(args + ["--conf", "spark.driver.memory=2g"])
+    submit_args = ' '.join([args_str, app_url, app_args])
     cmd = 'dcos --log-level=DEBUG spark --verbose run --submit-args="{0}"'.format(submit_args)
+
     LOGGER.info("Running {}".format(cmd))
     stdout = subprocess.check_output(cmd, shell=True).decode('utf-8')
+
     LOGGER.info("stdout: {}".format(stdout))
 
     regex = r"Submission id: (\S+)"