Merge pull request #13 from riptano/DSP-14970

DSP-14970 update to 0.8.0

Author: jtgrabowski

.DS_Store

@@ -0,0 +1,22 @@
+**Used Spark version**
+
+**Used Spark Job Server version**
+(Released version, git branch or docker image version)
+
+**Deployed mode**
+(client/cluster on Spark Standalone/YARN/Mesos/EMR or default)
+
+**Actual (wrong) behavior**
+
+**Steps to reproduce**
+
+**Logs**
+
+```
+some log
+```
+
+or as attached file (see below)
+
+
+Unused parts of this template should be removed (including this line).

.github/issue_template.md

@@ -9,6 +9,7 @@ test-reports/
 .idea*
 .vagrant
 *.pyc
+.DS_Store
 
 # ignore deployment configs
 config/*.conf

.gitignore

@@ -24,7 +24,7 @@ RUN sbt update
 # add the rest of the code
 COPY . .
 
-ENV SPARK_HOME /tmp/spark-2.1.0-bin-hadoop2.7
+ENV SPARK_HOME /tmp/spark-2.2.0-bin-hadoop2.7
 ENV JAVA_OPTIONS "-Xmx1500m -XX:MaxPermSize=512m -Dakka.test.timefactor=3"
 
 CMD ["/usr/src/app/run_tests.sh"]

Dockerfile.test

@@ -1,6 +1,6 @@
 #!/bin/bash
 # Script to start the job manager
-# args: <work dir for context> <cluster address> [proxy_user]
+# args: <master> <deployMode> <akkaAdress> <actorName> <workDir> [<proxyUser>]
 set -e
 
 get_abs_script_path() {
@@ -13,12 +13,8 @@ get_abs_script_path
 
 . $appdir/setenv.sh
 
-# Override logging options to provide per-context logging
-LOGGING_OPTS="-Dlog4j.configuration=file:$appdir/log4j-server.properties
-              -DLOG_DIR=$2"
-
 GC_OPTS="-XX:+UseConcMarkSweepGC
-         -verbose:gc -XX:+PrintGCTimeStamps -Xloggc:$appdir/gc.out
+         -verbose:gc -XX:+PrintGCTimeStamps
          -XX:MaxPermSize=512m
          -XX:+CMSClassUnloadingEnabled "
 
@@ -27,26 +23,46 @@ JAVA_OPTS="-XX:MaxDirectMemorySize=$MAX_DIRECT_MEMORY
 
 MAIN="spark.jobserver.JobManager"
 
-MESOS_OPTS=""
-if [ $1 == "mesos-cluster" ]; then
-    MESOS_OPTS="--master $MESOS_SPARK_DISPATCHER --deploy-mode cluster"
-    appdir=$REMOTE_JOBSERVER_DIR
-fi
+# copy files via spark-submit and read them from current (container) dir
+if [ $2 = "cluster" -a -z "$REMOTE_JOBSERVER_DIR" ]; then
+  SPARK_SUBMIT_OPTIONS="$SPARK_SUBMIT_OPTIONS
+    --master $1 --deploy-mode cluster
+    --conf spark.yarn.submit.waitAppCompletion=false
+    --files $appdir/log4j-cluster.properties,$conffile"
+  JAR_FILE="$appdir/spark-job-server.jar"
+  CONF_FILE=$(basename $conffile)
+  LOGGING_OPTS="-Dlog4j.configuration=log4j-cluster.properties"
+
+# use files in REMOTE_JOBSERVER_DIR
+elif [ $2 == "cluster" ]; then
+  SPARK_SUBMIT_OPTIONS="$SPARK_SUBMIT_OPTIONS
+    --master $1 --deploy-mode cluster
+    --conf spark.yarn.submit.waitAppCompletion=false"
+  JAR_FILE="$REMOTE_JOBSERVER_DIR/spark-job-server.jar"
+  CONF_FILE="$REMOTE_JOBSERVER_DIR/$(basename $conffile)"
+  LOGGING_OPTS="-Dlog4j.configuration=$REMOTE_JOBSERVER_DIR/log4j-cluster.properties"
 
-if [ ! -z $5 ]; then
-  cmd='$SPARK_HOME/bin/spark-submit --class $MAIN --driver-memory $JOBSERVER_MEMORY
-  --conf "spark.executor.extraJavaOptions=$LOGGING_OPTS"
-  --proxy-user $5
-  $MESOS_OPTS
-  --driver-java-options "$GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES"
-  $appdir/spark-job-server.jar $2 $3 $4 $conffile'
+# client mode, use files from app dir
 else
-  cmd='$SPARK_HOME/bin/spark-submit --class $MAIN --driver-memory $JOBSERVER_MEMORY
-  --conf "spark.executor.extraJavaOptions=$LOGGING_OPTS"
-  --driver-java-options "$GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES"
-  $MESOS_OPTS
-  $appdir/spark-job-server.jar $2 $3 $4 $conffile'
+  JAR_FILE="$appdir/spark-job-server.jar"
+  CONF_FILE="$conffile"
+  LOGGING_OPTS="-Dlog4j.configuration=file:$appdir/log4j-server.properties -DLOG_DIR=$5"
+  GC_OPTS="$GC_OPTS -Xloggc:$5/gc.out"
+fi
+
+if [ -n "$6" ]; then
+  SPARK_SUBMIT_OPTIONS="$SPARK_SUBMIT_OPTIONS --proxy-user $6"
 fi
 
-eval $cmd > /dev/null 2>&1 &
-# exec java -cp $CLASSPATH $GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES $MAIN $1 $2 $conffile 2>&1 &
+if [ -n "$JOBSERVER_KEYTAB" ]; then
+  SPARK_SUBMIT_OPTIONS="$SPARK_SUBMIT_OPTIONS --keytab $JOBSERVER_KEYTAB"
+fi
+
+cmd='$SPARK_HOME/bin/spark-submit --class $MAIN --driver-memory $JOBSERVER_MEMORY
+      --conf "spark.executor.extraJavaOptions=$LOGGING_OPTS"
+      $SPARK_SUBMIT_OPTIONS
+      --driver-java-options "$GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES $SPARK_SUBMIT_JAVA_OPTIONS"
+      $JAR_FILE $3 $4 $CONF_FILE'
+
+eval $cmd 2>&1 > $5/spark-job-server.out
+

README.md

@@ -19,7 +19,7 @@ lazy val akkaApp = Project(id = "akka-app", base = file("akka-app"))
 lazy val jobServer = Project(id = "job-server", base = file("job-server"))
   .settings(commonSettings)
   .settings(revolverSettings)
-  .settings(Assembly.settings)
+  .settings(assembly := null.asInstanceOf[File])
   .settings(
     description := "Spark as a Service: a RESTful job server for Apache Spark",
     libraryDependencies ++= sparkDeps ++ slickDeps ++ cassandraDeps ++ securityDeps ++ coreTestDeps,
@@ -35,7 +35,6 @@ lazy val jobServer = Project(id = "job-server", base = file("job-server"))
     fullClasspath in Compile <<= (fullClasspath in Compile).map { classpath =>
       extraJarPaths ++ classpath
     },
-    test in assembly := {},
     fork in Test := true
   )
   .settings(noPublishSettings)
@@ -48,9 +47,11 @@ lazy val jobServerTestJar = Project(id = "job-server-tests", base = file("job-se
   .settings(noPublishSettings)
   .dependsOn(jobServerApi)
   .disablePlugins(SbtScalariform)
+  .disablePlugins(ScoverageSbtPlugin) // do not include in coverage report
 
 lazy val jobServerApi = Project(id = "job-server-api", base = file("job-server-api"))
   .settings(commonSettings)
+  .settings(jobServerApiSettings)
   .settings(publishSettings)
   .disablePlugins(SbtScalariform)
 
@@ -93,15 +94,17 @@ lazy val root = Project(id = "root", base = file("."))
 lazy val jobServerExtrasSettings = revolverSettings ++ Assembly.settings ++ publishSettings ++ Seq(
   libraryDependencies ++= sparkExtraDeps ++ sparkExtraDepsTest,
   // Extras packages up its own jar for testing itself
-  test in Test <<= (test in Test).dependsOn(packageBin in Compile)
-    .dependsOn(clean in Compile),
+  test in Test <<= (test in Test).dependsOn(packageBin in Compile),
   fork in Test := true,
+  parallelExecution in Test := false,
   // Temporarily disable test for assembly builds so folks can package and get started.  Some tests
   // are flaky in extras esp involving paths.
   test in assembly := {},
   exportJars := true
 )
 
+lazy val jobServerApiSettings = Seq(libraryDependencies ++= sparkDeps ++ sparkExtraDeps)
+
 lazy val testPython = taskKey[Unit]("Launch a sub process to run the Python tests")
 lazy val buildPython = taskKey[Unit]("Build the python side of python support into an egg")
 lazy val buildPyExamples = taskKey[Unit]("Build the examples of python jobs into an egg")
@@ -138,7 +141,10 @@ lazy val dockerSettings = Seq(
     val sparkBuild = s"spark-${Versions.spark}"
     val sparkBuildCmd = scalaBinaryVersion.value match {
       case "2.11" =>
-        "./make-distribution.sh -Dscala-2.11 -Phadoop-2.7 -Phive"
+        Versions.spark match {
+          case s if s.startsWith("1") => {"./make-distribution.sh -Dscala-2.11 -Phadoop-2.7 -Phive"}
+          case _ => {"./dev/make-distribution.sh -Dscala-2.11 -Phadoop-2.7 -Phive"}
+        }
       case other => throw new RuntimeException(s"Scala version $other is not supported!")
     }
 
@@ -282,7 +288,7 @@ lazy val commonSettings = Defaults.coreDefaultSettings ++ dirSettings ++ implici
 
 lazy val scoverageSettings = {
   // Semicolon-separated list of regexs matching classes to exclude
-  coverageExcludedPackages := ".+Benchmark.*"
+  coverageExcludedPackages := ".+Benchmark.*;.+Example.*;.+TestJob"
 }
 
 /** Used for publishing `extras`, `api` and `python` jars. Main Spark Job Server assembly is published

bin/manager_start.sh

@@ -1,4 +1,4 @@
 #!/usr/bin/env bash
 set -e
-curl -L -o /tmp/spark.tgz http://d3kbcqa49mib13.cloudfront.net/spark-2.1.0-bin-hadoop2.7.tgz
+curl -L -o /tmp/spark.tgz http://d3kbcqa49mib13.cloudfront.net/spark-2.2.0-bin-hadoop2.7.tgz
 tar -xvzf /tmp/spark.tgz -C /tmp

build.sbt

@@ -1,5 +1,7 @@
 ## Step by step instruction on how to run Spark Job Server on EMR 4.2.0 (Spark 1.6.0)
 
+See also running in [cluster mode](cluster.md), running [YARN in client mode](yarn.md) and running on [Mesos](Mesos.md).
+
 ### Create EMR 4.2.0 cluster
 
 Create EMR cluster using AWS EMR console or aws cli.

ci/install-spark.sh

@@ -0,0 +1,66 @@
+## Configuring Job Server for YARN cluster mode
+
+See also running [YARN in client mode](yarn.md), running [YARN on EMR](EMR.md) and running on [Mesos](mesos.md).
+
+### Job Server configuration
+
+Add the following properties in your job server config file:
+- set `spark.master` property to `yarn`, `spark://...` or `mesos://...`
+- set `spark.submit.deployMode` property to `cluster`
+- set `spark.jobserver.context-per-jvm` to `true`
+- set `akka.remote.netty.tcp.hostname` to the cluster interface of the host running the frontend
+- set `akka.remote.netty.tcp.maximum-frame-size` to support big remote jars fetch
+
+Optional / required in spark standalone mode:
+- set `REMOTE_JOBSERVER_DIR` to `hdfs://...`, `file://...` or `http://...` in your settings `xxx.sh`
+- copy `spark-job-server.jar`, your job server config and `log4j-cluster.properties` file into this location
+
+Example job server config (replace `CLUSTER-IP` with the internal IP of the host running the job server frontend):
+
+    spark {
+      # deploy in yarn cluster mode
+      master = yarn
+      submit.deployMode = cluster
+
+      jobserver {
+        context-per-jvm = true
+
+        # start a H2 DB server, reachable in your cluster
+        sqldao {
+          jdbc {
+            url = "jdbc:h2:tcp://CLUSTER-IP:9092/h2-db;AUTO_RECONNECT=TRUE"
+          }
+        }
+        startH2Server = false
+      }
+    }
+
+    # start akka on this interface, reachable from your cluster
+    akka {
+      remote.netty.tcp {
+        hostname = "CLUSTER-IP"
+
+        # This controls the maximum message size, including job results, that can be sent
+        maximum-frame-size = 100 MiB
+      }
+    }
+
+Note:
+- YARN transfers the files provided via `--files` submit option into the cluster / container. Spark standalone does not support this in cluster mode and you have to transfer them manual.
+- Instead of running a H2 DB instance you can also run a real DB reachable inside your cluster. You can't use the default (host only) H2 configuration in a cluster setup.
+- Akka binds by [default](../job-server/src/main/resources/application.conf) to the local host interface and is not reachable from the cluster. You need to configure the akka hostname to the cluster internal address.
+
+### Reading files uploaded via frontend
+
+Files uploaded via the data API (`/data`) are stored on your job server frontend host.
+Call the [DataFileCache](../job-server-api/src/main/scala/spark/jobserver/api/SparkJobBase.scala) API implemented by the job environment in your spark jobs to access them:
+
+```scala
+  object RemoteDriverExample extends NewSparkJob {
+    def runJob(sc: SparkContext, runtime: JobEnvironment, data: JobData): JobOutput =
+      runtime.getDataFile(...)
+```
+
+The job server transfers the files via akka to the host running your driver and caches them there.
+
+Note: Files uploaded via the JAR or binary API are stored and transfered via the Job DB.