Add support for pyspark streaming applications

PNDA-2729

Author: jeclarke

CHANGELOG.md

@@ -2,9 +2,12 @@
 All notable changes to this project will be documented in this file.
 
 ## Unreleased
+### Added
+ - PNDA-2729: Added support for spark streaming jobs written in python (pyspark). Use `main_py` instead of `main_jar` in properties.json and specify additional files using `py_files`.
 ### Changed
  - PNDA-2700: Spark streaming jobs no longer require upstart.conf or yarn-kill.py files, default ones are supplied by the deployment manager.
 
+
 ## [0.3.0] 2017-01-20
 ### Fixed
 - PNDA-2498: Application package data is now stored in HDFS with a reference to the path only held in the HBase record. This prevents HBase being overloaded with large packages (100MB+).

api/src/main/resources/plugins/sparkStreaming.py

@@ -80,26 +80,34 @@ def create_component(self, staged_component_path, application_name, component, p
 
         if 'component_spark_submit_args' not in properties:
             properties['component_spark_submit_args'] = ''
+        if 'component_py_files' not in properties:
+            properties['component_py_files'] = ''
 
         if 'upstart.conf' in component['component_detail']:
             # old style applications for backward compatibility
             service_script = 'upstart.conf'
             service_script_install_path = '/etc/init/%s.conf' % service_name
         else:
             # new style applications that don't need to provide upstart.conf or yarn-kill.py
-            if 'component_main_class' not in properties:
+            if 'component_main_jar' in properties and 'component_main_class' not in properties:
                 raise Exception('properties.json must contain "main_class" for %s sparkStreaming %s' % (application_name, component['component_name']))
-            if 'component_main_jar' not in properties:
-                raise Exception('properties.json must contain "main_jar" for %s sparkStreaming %s' % (application_name, component['component_name']))
+
+            java_app = None
+            if 'component_main_jar' in properties:
+                java_app = True
+            elif 'component_main_py' in properties:
+                java_app = False
+            else:
+                raise Exception('properties.json must contain "main_jar or main_py" for %s sparkStreaming %s' % (application_name, component['component_name']))
 
             this_dir = os.path.dirname(os.path.realpath(__file__))
             copy(os.path.join(this_dir, 'yarn-kill.py'), staged_component_path)
             distro = platform.dist()
             if redhat:
-                service_script = 'systemd.service.tpl'
+                service_script = 'systemd.service.tpl' if java_app else 'systemd.service.py.tpl'
                 service_script_install_path = '/usr/lib/systemd/system/%s.service' % service_name
             else:
-                service_script = 'upstart.conf.tpl'
+                service_script = 'upstart.conf.tpl' if java_app else 'upstart.conf.py.tpl'
                 service_script_install_path = '/etc/init/%s.conf' % service_name
             copy(os.path.join(this_dir, service_script), staged_component_path)
 

api/src/main/resources/plugins/systemd.service.py.tpl

@@ -0,0 +1,12 @@
+[Unit]
+Description=PNDA Application: ${component_application}-${component_name}
+
+[Service]
+Type=simple
+User=hdfs
+WorkingDirectory=/opt/${environment_namespace}/${component_application}/${component_name}/
+ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+ExecStopPost=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+ExecStart=/usr/bin/spark-submit --driver-java-options "-Dlog4j.configuration=file:////opt/${environment_namespace}/${component_application}/${component_name}/log4j.properties" --conf 'spark.executor.extraJavaOptions=-Dlog4j.configuration=file:////opt/${environment_namespace}/${component_application}/${component_name}/log4j.properties' --name '${component_job_name}' --master yarn-cluster --py-files application.properties,${component_py_files} ${component_spark_submit_args} ${component_main_py}
+Restart=always
+RestartSec=2

api/src/main/resources/plugins/upstart.conf.py.tpl

@@ -0,0 +1,9 @@
+start on runlevel [2345]
+stop on runlevel [016]
+respawn
+respawn limit unlimited
+pre-start exec /opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+pre-stop exec /opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+env programDir=/opt/${environment_namespace}/${component_application}/${component_name}/
+chdir /opt/${environment_namespace}/${component_application}/${component_name}/
+exec sudo -u hdfs spark-submit --driver-java-options "-Dlog4j.configuration=file:///${programDir}log4j.properties" --conf 'spark.executor.extraJavaOptions=-Dlog4j.configuration=file:///${programDir}log4j.properties' --name '${component_job_name}' --master yarn-cluster --py-files application.properties,${component_py_files} ${component_spark_submit_args} ${component_main_py}

api/src/main/resources/plugins/upstart.conf.tpl

@@ -5,4 +5,5 @@ respawn limit unlimited
 pre-start exec /opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
 pre-stop exec /opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
 env programDir=/opt/${environment_namespace}/${component_application}/${component_name}/
-exec sudo -u hdfs spark-submit --driver-java-options "-Dlog4j.configuration=file:///${programDir}log4j.properties" --class ${component_main_class} --name '${component_job_name}' --master yarn-cluster --files ${programDir}log4j.properties ${component_spark_submit_args} ${programDir}${component_main_jar}
+chdir /opt/${environment_namespace}/${component_application}/${component_name}/
+exec sudo -u hdfs spark-submit --driver-java-options "-Dlog4j.configuration=file:///${programDir}log4j.properties" --class ${component_main_class} --name '${component_job_name}' --master yarn-cluster --files log4j.properties ${component_spark_submit_args} ${component_main_jar}