Python Bindings for launching PySpark Jobs from the JVM

README.md

@@ -24,6 +24,7 @@ We've been asked by an Apache Spark Committer to work outside of the Apache infr
 
 This is a collaborative effort by several folks from different companies who are interested in seeing this feature be successful.  Companies active in this project include (alphabetically):
 
+- Bloomberg
 - Google
 - Haiwen
 - Hyperpilot

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -335,8 +335,8 @@ object SparkSubmit {
     (clusterManager, deployMode) match {
       case (KUBERNETES, CLIENT) =>
         printErrorAndExit("Client mode is currently not supported for Kubernetes.")
-      case (KUBERNETES, CLUSTER) if args.isPython || args.isR =>
-        printErrorAndExit("Kubernetes does not currently support python or R applications.")
+      case (KUBERNETES, CLUSTER) if args.isR =>
+        printErrorAndExit("Kubernetes does not currently support R applications.")
       case (STANDALONE, CLUSTER) if args.isPython =>
         printErrorAndExit("Cluster deploy mode is currently not supported for python " +
           "applications on standalone clusters.")
@@ -620,9 +620,23 @@ object SparkSubmit {
 
     if (isKubernetesCluster) {
       childMainClass = "org.apache.spark.deploy.kubernetes.submit.Client"
-      childArgs += args.primaryResource
-      childArgs += args.mainClass
-      childArgs ++= args.childArgs
+      if (args.isPython) {
+        childArgs += "--py-file"
+        childArgs += args.primaryResource
+        childArgs += "--main-class"
+        childArgs += "org.apache.spark.deploy.PythonRunner"
+        childArgs += "--other-py-files"
+        childArgs += args.pyFiles
+      } else {
+        childArgs += "--primary-java-resource"
+        childArgs += args.primaryResource
+        childArgs += "--main-class"
+        childArgs += args.mainClass
+      }
+      args.childArgs.foreach { arg =>
+        childArgs += "--arg"
+        childArgs += arg
+      }
     }
 
     // Load any properties specified through --conf and the default properties file

docs/running-on-kubernetes.md

@@ -180,6 +180,32 @@ The above mechanism using `kubectl proxy` can be used when we have authenticatio
 kubernetes-client library does not support. Authentication using X509 Client Certs and OAuth tokens
 is currently supported.
 
+### Running PySpark
+
+Running PySpark on Kubernetes leverages the same spark-submit logic when launching on Yarn and Mesos. 
+Python files can be distributed by including, in the conf, `--py-files` 
+
+Below is an example submission: 
+
+
+```
+    bin/spark-submit \
+      --deploy-mode cluster \
+      --master k8s://http://127.0.0.1:8001 \
+      --kubernetes-namespace default \
+      --conf spark.executor.memory=500m \
+      --conf spark.driver.memory=1G \
+      --conf spark.driver.cores=1 \
+      --conf spark.executor.cores=1 \
+      --conf spark.executor.instances=1 \
+      --conf spark.app.name=spark-pi \
+      --conf spark.kubernetes.driver.docker.image=spark-driver-py:latest \
+      --conf spark.kubernetes.executor.docker.image=spark-executor-py:latest \
+      --conf spark.kubernetes.initcontainer.docker.image=spark-init:latest \
+      --py-files local:///opt/spark/examples/src/main/python/sort.py \
+      local:///opt/spark/examples/src/main/python/pi.py 100
+```
+
 ## Dynamic Executor Scaling
 
 Spark on Kubernetes supports Dynamic Allocation with cluster mode. This mode requires running

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/InitContainerResourceStagingServerSecretPlugin.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.spark.deploy.kubernetes
 
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder, Secret}
+import io.fabric8.kubernetes.api.model.{Container, ContainerBuilder, Pod, PodBuilder, Secret}
 
 import org.apache.spark.deploy.kubernetes.constants._
 
@@ -27,13 +27,13 @@ private[spark] trait InitContainerResourceStagingServerSecretPlugin {
    * from a resource staging server.
    */
   def mountResourceStagingServerSecretIntoInitContainer(
-      initContainer: ContainerBuilder): ContainerBuilder
+      initContainer: Container): Container
 
   /**
    * Configure the pod to attach a Secret volume which hosts secret files allowing the
    * init-container to retrieve dependencies from the resource staging server.
    */
-  def addResourceStagingServerSecretVolumeToPod(basePod: PodBuilder): PodBuilder
+  def addResourceStagingServerSecretVolumeToPod(basePod: Pod): Pod
 }
 
 private[spark] class InitContainerResourceStagingServerSecretPluginImpl(
@@ -42,21 +42,25 @@ private[spark] class InitContainerResourceStagingServerSecretPluginImpl(
     extends InitContainerResourceStagingServerSecretPlugin {
 
   override def mountResourceStagingServerSecretIntoInitContainer(
-      initContainer: ContainerBuilder): ContainerBuilder = {
-    initContainer.addNewVolumeMount()
-      .withName(INIT_CONTAINER_SECRET_VOLUME_NAME)
-      .withMountPath(initContainerSecretMountPath)
-      .endVolumeMount()
+      initContainer: Container): Container = {
+    new ContainerBuilder(initContainer)
+      .addNewVolumeMount()
+        .withName(INIT_CONTAINER_SECRET_VOLUME_NAME)
+        .withMountPath(initContainerSecretMountPath)
+        .endVolumeMount()
+      .build()
   }
 
-  override def addResourceStagingServerSecretVolumeToPod(basePod: PodBuilder): PodBuilder = {
-    basePod.editSpec()
-      .addNewVolume()
-        .withName(INIT_CONTAINER_SECRET_VOLUME_NAME)
-        .withNewSecret()
-          .withSecretName(initContainerSecretName)
-          .endSecret()
-        .endVolume()
-      .endSpec()
+  override def addResourceStagingServerSecretVolumeToPod(basePod: Pod): Pod = {
+    new PodBuilder(basePod)
+      .editSpec()
+        .addNewVolume()
+          .withName(INIT_CONTAINER_SECRET_VOLUME_NAME)
+          .withNewSecret()
+            .withSecretName(initContainerSecretName)
+            .endSecret()
+          .endVolume()
+        .endSpec()
+      .build()
   }
 }

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/PodWithDetachedInitContainer.scala

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.deploy.kubernetes
+
+import io.fabric8.kubernetes.api.model.{Container, Pod}
+
+private[spark] case class PodWithDetachedInitContainer(
+    pod: Pod,
+    initContainer: Container,
+    mainContainer: Container)

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/SparkPodInitContainerBootstrap.scala

@@ -19,19 +19,25 @@ package org.apache.spark.deploy.kubernetes
 import io.fabric8.kubernetes.api.model.{ContainerBuilder, EmptyDirVolumeSource, PodBuilder, VolumeMount, VolumeMountBuilder}
 
 import org.apache.spark.deploy.kubernetes.constants._
-import org.apache.spark.deploy.kubernetes.submit.{ContainerNameEqualityPredicate, InitContainerUtil}
 
+/**
+ * This is separated out from the init-container steps API because this component can be reused to
+ * set up the init-container for executors as well.
+ */
 private[spark] trait SparkPodInitContainerBootstrap {
   /**
    * Bootstraps an init-container that downloads dependencies to be used by a main container.
    * Note that this primarily assumes that the init-container's configuration is being provided
    * by a ConfigMap that was installed by some other component; that is, the implementation
    * here makes no assumptions about how the init-container is specifically configured. For
    * example, this class is unaware if the init-container is fetching remote dependencies or if
-   * it is fetching dependencies from a resource staging server.
+   * it is fetching dependencies from a resource staging server. Additionally, the container itself
+   * is not actually attached to the pod, but the init container is returned so it can be attached
+   * by InitContainerUtil after the caller has decided to make any changes to it.
    */
   def bootstrapInitContainerAndVolumes(
-      mainContainerName: String, originalPodSpec: PodBuilder): PodBuilder
+      originalPodWithUnattachedInitContainer: PodWithDetachedInitContainer)
+      : PodWithDetachedInitContainer
 }
 
 private[spark] class SparkPodInitContainerBootstrapImpl(
@@ -41,13 +47,12 @@ private[spark] class SparkPodInitContainerBootstrapImpl(
     filesDownloadPath: String,
     downloadTimeoutMinutes: Long,
     initContainerConfigMapName: String,
-    initContainerConfigMapKey: String,
-    resourceStagingServerSecretPlugin: Option[InitContainerResourceStagingServerSecretPlugin])
+    initContainerConfigMapKey: String)
   extends SparkPodInitContainerBootstrap {
 
   override def bootstrapInitContainerAndVolumes(
-      mainContainerName: String,
-      originalPodSpec: PodBuilder): PodBuilder = {
+      originalPodWithUnattachedInitContainer: PodWithDetachedInitContainer)
+      : PodWithDetachedInitContainer = {
     val sharedVolumeMounts = Seq[VolumeMount](
       new VolumeMountBuilder()
         .withName(INIT_CONTAINER_DOWNLOAD_JARS_VOLUME_NAME)
@@ -58,7 +63,7 @@ private[spark] class SparkPodInitContainerBootstrapImpl(
         .withMountPath(filesDownloadPath)
         .build())
 
-    val initContainer = new ContainerBuilder()
+    val initContainer = new ContainerBuilder(originalPodWithUnattachedInitContainer.initContainer)
       .withName(s"spark-init")
       .withImage(initContainerImage)
       .withImagePullPolicy(dockerImagePullPolicy)
@@ -68,11 +73,8 @@ private[spark] class SparkPodInitContainerBootstrapImpl(
         .endVolumeMount()
       .addToVolumeMounts(sharedVolumeMounts: _*)
       .addToArgs(INIT_CONTAINER_PROPERTIES_FILE_PATH)
-    val resolvedInitContainer = resourceStagingServerSecretPlugin.map { plugin =>
-      plugin.mountResourceStagingServerSecretIntoInitContainer(initContainer)
-    }.getOrElse(initContainer).build()
-    val podWithBasicVolumes = InitContainerUtil.appendInitContainer(
-        originalPodSpec, resolvedInitContainer)
+      .build()
+    val podWithBasicVolumes = new PodBuilder(originalPodWithUnattachedInitContainer.pod)
       .editSpec()
         .addNewVolume()
           .withName(INIT_CONTAINER_PROPERTIES_FILE_VOLUME)
@@ -92,17 +94,20 @@ private[spark] class SparkPodInitContainerBootstrapImpl(
           .withName(INIT_CONTAINER_DOWNLOAD_FILES_VOLUME_NAME)
           .withEmptyDir(new EmptyDirVolumeSource())
           .endVolume()
-        .editMatchingContainer(new ContainerNameEqualityPredicate(mainContainerName))
-          .addToVolumeMounts(sharedVolumeMounts: _*)
-          .addNewEnv()
-            .withName(ENV_MOUNTED_FILES_DIR)
-            .withValue(filesDownloadPath)
-            .endEnv()
-          .endContainer()
         .endSpec()
-    resourceStagingServerSecretPlugin.map { plugin =>
-      plugin.addResourceStagingServerSecretVolumeToPod(podWithBasicVolumes)
-    }.getOrElse(podWithBasicVolumes)
+      .build()
+    val mainContainerWithMountedFiles = new ContainerBuilder(
+      originalPodWithUnattachedInitContainer.mainContainer)
+        .addToVolumeMounts(sharedVolumeMounts: _*)
+        .addNewEnv()
+          .withName(ENV_MOUNTED_FILES_DIR)
+          .withValue(filesDownloadPath)
+          .endEnv()
+        .build()
+    PodWithDetachedInitContainer(
+      podWithBasicVolumes,
+      initContainer,
+      mainContainerWithMountedFiles)
   }
 
 }

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala

@@ -418,6 +418,14 @@ package object config extends Logging {
       .stringConf
       .createOptional
 
+  private[spark] val INIT_CONTAINER_REMOTE_PYSPARK_FILES =
+    ConfigBuilder("spark.kubernetes.initcontainer.remotePyFiles")
+      .doc("Comma-separated list of Python file URIs to download in the init-container. This is" +
+        " calculated given the list of python files sent to spark-submit.")
+      .internal()
+      .stringConf
+      .createOptional
+
   private[spark] val INIT_CONTAINER_DOCKER_IMAGE =
     ConfigBuilder("spark.kubernetes.initcontainer.docker.image")
       .doc("Image for the driver and executor's init-container that downloads dependencies.")

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala

@@ -67,6 +67,8 @@ package object constants {
   private[spark] val ENV_DRIVER_ARGS = "SPARK_DRIVER_ARGS"
   private[spark] val ENV_DRIVER_JAVA_OPTS = "SPARK_DRIVER_JAVA_OPTS"
   private[spark] val ENV_MOUNTED_FILES_DIR = "SPARK_MOUNTED_FILES_DIR"
+  private[spark] val ENV_PYSPARK_FILES = "PYSPARK_FILES"
+  private[spark] val ENV_PYSPARK_PRIMARY = "PYSPARK_PRIMARY"
 
   // Bootstrapping dependencies with the init-container
   private[spark] val INIT_CONTAINER_ANNOTATION = "pod.beta.kubernetes.io/init-containers"