Support Adding Application Egg to PySpark (spark-jobserver#941)

* add submitted egg to spark context on subprocess start

* slight revision

* test for empty and nonexistent strings

Author: rfrohock

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

@@ -64,6 +64,7 @@ case class PythonJob[X <: PythonContextLike](eggPath: String,
         Process(
           Seq(sc.pythonExecutable, "-m", "sparkjobserver.subprocess", server.getListeningPort.toString),
           None,
+          "EGGPATH" -> eggPath,
           "PYTHONPATH" -> pythonPath,
           "PYSPARK_PYTHON" -> sc.pythonExecutable)
       val err = new StringBuffer

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

@@ -16,6 +16,7 @@
 
 from __future__ import print_function
 import sys
+import os
 from importlib import import_module
 from py4j.java_gateway import JavaGateway, java_import, GatewayClient
 from pyhocon import ConfigFactory
@@ -71,6 +72,7 @@ def import_class(cls):
     spark_conf = SparkConf(_jconf=jspark_conf)
     context_class = jcontext.contextType()
     context = None
+    sc = None
     if context_class == 'org.apache.spark.api.java.JavaSparkContext':
         context = SparkContext(
                 gateway=gateway, jsc=jcontext, conf=spark_conf)
@@ -98,6 +100,15 @@ def import_class(cls):
             exit_with_failure(
                     "Expected JavaSparkContext, SQLContext "
                     "or HiveContext but received %s" % repr(context_class), 2)
+
+    egg_path = os.environ.get("EGGPATH", None)
+    if egg_path and sc:
+        try:
+            sc.addPyFile(egg_path)
+        except Exception as error:
+            exit_with_failure(
+                "Error while adding Python Egg to Spark Context: %s\n%s" %
+                (repr(error), traceback.format_exc()), 5)
     try:
         job_data = job.validate(context, None, job_config)
     except Exception as error: