Update Versions for Spark 2.4

Upgrade Py4j To the Correct Version

Spark 2.4 Pyspark Compatiblity

Several version bumps of associated libraries. Also includes
converting the current subprocess.py and associated PythonJob
to use an authentication token when starting Pyspark. The method
used for generating the token is the same as inside native Spark. The
token is passed via environment variable from the PythonJob class to
the subprocess.py process which starts up the python interpreter.

Fix Tests

Switches warehouse directories per test job since the "DROP TABLE"
doesn't seem to work in SPARK 2.4 like it did in Spark 2.2. Extend
the timeout on the python Job spec.

Author: RussellSpitzer

build.sbt

@@ -297,7 +297,8 @@ lazy val publishSettings = Seq(
   autoScalaLibrary := false,
   credentials += Credentials(Path.userHome / ".sbt" / ".credentials"),
   publishMavenStyle := true,
-  publishTo := Some(sys.env("MVN_PUBLISH_REPO") at sys.env("MVN_PUBLISH_URL")),
+  publishTo := Some(sys.env.getOrElse("MVN_PUBLISH_REPO", "NONE")
+    at sys.env.getOrElse("MVN_PUBLISH_URL", "NONE")),
   licenses += ("Apache-2.0", url("http://choosealicense.com/licenses/apache/")),
   pomIncludeRepository := { _ => false },
   /** Since users are encouraged to use dse-spark-dependencies, which provides most of the needed

job-server-extras/src/main/scala/spark/jobserver/context/SessionContextFactory.scala

@@ -1,5 +1,8 @@
 package spark.jobserver.context
 
+import java.io.File
+import java.nio.file.Files
+
 import com.typesafe.config.Config
 import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.sql.SparkSession
@@ -19,6 +22,13 @@ class SessionContextFactory extends ScalaContextFactory {
 
   def isValidJob(job: SparkJobBase): Boolean = job.isInstanceOf[SparkSessionJob]
 
+  // creates a stub warehouse dir for derby/hive metastore
+  def makeWarehouseDir(): File = {
+    val warehouseDir = Files.createTempDirectory("warehouse").toFile()
+    warehouseDir.delete()
+    warehouseDir
+  }
+
   def makeContext(sparkConf: SparkConf, config: Config, contextName: String): C = {
     val builder = SparkSession.builder()
     builder.config(sparkConf).appName(contextName)

job-server-extras/src/main/scala/spark/jobserver/python/PythonJob.scala

@@ -1,12 +1,15 @@
 package spark.jobserver.python
 
+import java.security.SecureRandom
+
 import com.typesafe.config.Config
-import org.scalactic.{Good, Every, Or}
+import org.apache.spark.SparkConf
+import org.scalactic.{Every, Good, Or}
 import org.slf4j.LoggerFactory
-import py4j.GatewayServer
-import spark.jobserver.api.{SparkJobBase, ValidationProblem, JobEnvironment}
+import spark.jobserver.api.{JobEnvironment, SparkJobBase, ValidationProblem}
+import org.spark_project.guava.hash.HashCodes
 
-import scala.sys.process.{ProcessLogger, Process}
+import scala.sys.process.{Process, ProcessLogger}
 import scala.util.{Failure, Success, Try}
 
 case class PythonJob[X <: PythonContextLike](eggPath: String,
@@ -41,6 +44,17 @@ case class PythonJob[X <: PythonContextLike](eggPath: String,
                         runtime: JobEnvironment,
                         config: Config): Or[Config, Every[ValidationProblem]] = Good(config)
 
+  /**
+    * Copied from Spark private method in 2.4.0
+    */
+  def createSecret(conf: SparkConf): String = {
+    val bits = conf.getInt("spark.authenticate.secretBitLength", 256)
+    val rnd = new SecureRandom()
+    val secretBytes = new Array[Byte](bits / java.lang.Byte.SIZE)
+    rnd.nextBytes(secretBytes)
+    HashCodes.fromBytes(secretBytes).toString()
+  }
+
   /**
     * This is the entry point for a Spark Job Server to execute Python jobs.
     * It calls a Python subprocess to execute the relevant Python Job class.
@@ -53,7 +67,12 @@ case class PythonJob[X <: PythonContextLike](eggPath: String,
   override def runJob(sc: X, runtime: JobEnvironment, data: Config): Any = {
     logger.info(s"Running $modulePath from $eggPath")
     val ep = endpoint(sc, runtime.contextConfig, runtime.jobId, data)
-    val server = new GatewayServer(ep, 0)
+    val secret = createSecret(sc.sparkContext.getConf)
+    val server = new py4j.GatewayServer.GatewayServerBuilder()
+      .entryPoint(ep)
+      .javaPort(0)
+      .authToken(secret)
+      .build()
     val pythonPathDelimiter : String = if (System.getProperty("os.name").indexOf("Win") >= 0) ";" else ":"
     val pythonPath = (eggPath +: sc.pythonPath).mkString(pythonPathDelimiter)
     logger.info(s"Using Python path of ${pythonPath}")
@@ -66,7 +85,8 @@ case class PythonJob[X <: PythonContextLike](eggPath: String,
           None,
           "EGGPATH" -> eggPath,
           "PYTHONPATH" -> pythonPath,
-          "PYSPARK_PYTHON" -> sc.pythonExecutable)
+          "PYSPARK_PYTHON" -> sc.pythonExecutable,
+          "PYSPARK_GATEWAY_SECRET" -> secret)
       val err = new StringBuffer
       val procLogger =
         ProcessLogger(

job-server-extras/src/test/scala/spark/jobserver/SessionJobSpec.scala

@@ -12,11 +12,12 @@ import scala.concurrent.duration._
 
 class TestSessionContextFactory extends SessionContextFactory {
 
-  override def makeContext(sparkConf: SparkConf, config: Config,  contextName: String): C = {
+  override def makeContext(sparkConf: SparkConf, config: Config, contextName: String): C = {
     val builder = SparkSession.builder()
     builder.config(sparkConf).appName(contextName).master("local")
     builder.config("javax.jdo.option.ConnectionURL", "jdbc:derby:memory:myDB;create=true")
     builder.config("javax.jdo.option.ConnectionDriverName", "org.apache.derby.jdbc.EmbeddedDriver")
+    builder.config("spark.sql.warehouse.dir", makeWarehouseDir().getAbsolutePath)
     try {
       builder.enableHiveSupport()
     } catch {

job-server-extras/src/test/scala/spark/jobserver/python/PythonJobManagerSpec.scala

@@ -39,7 +39,7 @@ class PythonJobManagerSpec extends ExtrasJobSpecBase(PythonJobManagerSpec.getNew
         "example_jobs.word_count.WordCountSparkSessionJob",
         ConfigFactory.parseString("""input.strings = ["a", "b", "a"]"""),
         errorEvents ++ syncEvents)
-      expectMsgPF(3 seconds, "Expected a JobResult or JobErroredOut message!") {
+      expectMsgPF(10 seconds, "Expected a JobResult or JobErroredOut message!") {
         case JobResult(_, x) => x should matchPattern {
           case m: java.util.Map[_, _] if m.asScala == Map("b" -> 1, "a" -> 2) =>
         }

job-server-python/src/python/run-tests.sh

@@ -1,5 +1,5 @@
 #!/usr/bin/env bash
-PYTHONPATH=.:$SPARK_HOME/python/lib/pyspark.zip:$SPARK_HOME/python/lib/py4j-0.10.4-src.zip:$PYTHONPATH python test/apitests.py
+PYTHONPATH=.:$SPARK_HOME/python/lib/pyspark.zip:$SPARK_HOME/python/lib/py4j-0.10.7-src.zip:$PYTHONPATH python test/apitests.py
 exitCode=$?
 #This sleep is here so that all of Spark's shutdown stdout if written before we exit,
 #so that we return cleanly to the command prompt.

job-server-python/src/python/setup.py

@@ -9,5 +9,5 @@
     url="https://github.com/spark-jobserver/spark-jobserver",
     license="Apache License 2.0",
     packages=find_packages(exclude=["test*", "example*"]),
-    install_requires=["pyhocon", "py4j"]
+    install_requires=["pyhocon", "py4j==0.10.7"]
 )

job-server-python/src/python/sparkjobserver/subprocess.py

@@ -18,7 +18,7 @@
 import sys
 import os
 from importlib import import_module
-from py4j.java_gateway import JavaGateway, java_import, GatewayClient
+from py4j.java_gateway import JavaGateway, java_import, GatewayClient, GatewayParameters
 from pyhocon import ConfigFactory
 from pyspark.context import SparkContext, SparkConf
 from pyspark.sql import SQLContext, HiveContext, SparkSession
@@ -53,7 +53,10 @@ def import_class(cls):
 
 if __name__ == "__main__":
     port = int(sys.argv[1])
-    gateway = JavaGateway(GatewayClient(port=port), auto_convert=True)
+
+    gateway_secret = os.environ["PYSPARK_GATEWAY_SECRET"]
+    parameters = GatewayParameters(port=port, auto_convert=True, auth_token=gateway_secret)
+    gateway = JavaGateway(gateway_parameters=parameters)
     entry_point = gateway.entry_point
     imports = entry_point.getPy4JImports()
     for i in imports:

job-server-python/src/test/scala/spark/jobserver/python/SubprocessSpec.scala

@@ -3,14 +3,15 @@ package spark.jobserver.python
 import java.io.File
 import java.nio.file.Files
 
-import com.typesafe.config.{ConfigRenderOptions, Config, ConfigFactory}
+import com.typesafe.config.{Config, ConfigFactory, ConfigRenderOptions}
 import org.apache.spark.SparkContext
 import org.apache.spark.api.java.JavaSparkContext
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.hive.HiveContext
 import org.apache.spark.{SparkConf, SparkContext}
-import org.scalatest.{BeforeAndAfterAll, Matchers, FunSpec}
+import org.scalatest.{BeforeAndAfterAll, FunSpec, Matchers}
 import py4j.GatewayServer
+import py4j.GatewayServer.GatewayServerBuilder
 
 import scala.collection.JavaConverters._
 import scala.sys.process.Process
@@ -34,10 +35,10 @@ object SubprocessSpec {
   lazy val jobServerPath = getPythonDir("src/python")
 
   lazy val pysparkPath = sys.env.get("SPARK_HOME").map(d => s"$d/python/lib/pyspark.zip")
-  lazy val py4jPath = sys.env.get("SPARK_HOME").map(d => s"$d/python/lib/py4j-0.10.4-src.zip")
+  lazy val py4jPath = sys.env.get("SPARK_HOME").map(d => s"$d/python/lib/py4j-0.10.7-src.zip")
   lazy val sparkPaths = sys.env.get("SPARK_HOME").map{sh =>
     val pysparkPath = s"$sh/python/lib/pyspark.zip"
-    val py4jPath = s"$sh/python/lib/py4j-0.10.4-src.zip"
+    val py4jPath = s"$sh/python/lib/py4j-0.10.7-src.zip"
     Seq(pysparkPath, py4jPath)
   }.getOrElse(Seq())
   lazy val originalPythonPath = sys.env.get("PYTHONPATH")
@@ -93,18 +94,28 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
     val pathList = Seq(jobServerPath) ++ sparkPaths ++ originalPythonPath.toSeq
     val p = pathList.mkString(pythonPathDelimiter)
     // Scarman 10-13-2016
-    //println(p)
+    println(p)
     p
   }
 
+  lazy val env = Seq (
+    "PYTHONPATH" -> pythonPath,
+    "PYSPARK_GATEWAY_SECRET" -> secret
+  )
+
   // creates a stub warehouse dir for derby/hive metastore
   def makeWarehouseDir(): File = {
     val warehouseDir = Files.createTempDirectory("warehouse").toFile()
     warehouseDir.delete()
     warehouseDir
   }
+  val secret = "Test"
   def buildGateway(endpoint: TestEndpoint): GatewayServer = {
-    val server = new GatewayServer(endpoint, 0)
+    val server = new GatewayServerBuilder()
+      .entryPoint(endpoint)
+      .javaPort(0)
+      .authToken(secret)
+      .build()
     //Server runs asynchronously on a dedicated thread. See Py4J source for more detail
     server.start()
     server
@@ -162,7 +173,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (0)
       endpoint.result should matchPattern {
@@ -181,7 +192,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (1)
       endpoint.validationProblems should be (Some(Seq("config input.strings not found")))
@@ -205,7 +216,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (0)
       endpoint.result should matchPattern {
@@ -235,7 +246,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (0)
       endpoint.result should matchPattern {
@@ -268,7 +279,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (0)
       endpoint.result should be ("done")
@@ -283,7 +294,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw2.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode2 = process2.!
       pythonExitCode2 should be (0)
       endpoint2.result should matchPattern {
@@ -308,7 +319,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (2)
       stopGateway(gw)
@@ -323,7 +334,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (4)
       stopGateway(gw)
@@ -338,7 +349,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (3)
       stopGateway(gw)
@@ -356,7 +367,7 @@ class SubprocessSpec extends FunSpec with Matchers with BeforeAndAfterAll {
         Process(
           Seq("python", "-m", "sparkjobserver.subprocess", gw.getListeningPort.toString),
           None,
-          "PYTHONPATH" -> pythonPath)
+          env: _*)
       val pythonExitCode = process.!
       pythonExitCode should be (0)
       endpoint.result should be ("Hello World 3")

project/Assembly.scala

@@ -10,7 +10,7 @@ object Assembly {
     assemblyExcludedJars in assembly <<= (fullClasspath in assembly) map { _ filter { cp =>
       List("servlet-api", "guice-all", "junit", "uuid",
         "jetty", "jsp-api-2.0", "antlr", "avro", "slf4j-log4j", "log4j-1.2",
-        "scala-actors", "commons-cli", "stax-api", "mockito",
+        "scala-actors", "commons-cli", "stax-api", "mockito", "lz4",
         // we rely on whatever version DSE has:
         "spark", "netty", "dse-java-driver").exists(cp.data.getName.startsWith(_))
     } },