Python Bindings for launching PySpark Jobs from the JVM (v1) #351
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| /* | ||
| * 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.submit | ||
|
|
||
| import io.fabric8.kubernetes.api.model.{Container, PodBuilder} | ||
|
|
||
| import org.apache.spark.deploy.kubernetes.constants._ | ||
|
|
||
| /** | ||
| * Trait that is responsible for providing full file-paths dynamically after | ||
| * the filesDownloadPath has been defined. The file-names are then stored in the | ||
| * environmental variables in the driver-pod. | ||
|
There was a problem hiding this comment. Given this description, can we use ContainerLocalizedFileResolver? There was a problem hiding this comment.
|
||
| */ | ||
| private[spark] trait DriverPodKubernetesFileMounter { | ||
|
There was a problem hiding this comment. With the V2 file staging server, do we need special code for staging python files? There was a problem hiding this comment. They're just added to There was a problem hiding this comment. I think there's a legitimate question of if we want these files to be deployed in the same location as where the |
||
| def addPySparkFiles(mainAppResource: String, pythonFiles: List[String], | ||
|
There was a problem hiding this comment. This seems to be python specific, but it's trait name is not. Should the trait name include There was a problem hiding this comment. This will be a more general file mounter for rfiles as well. There was a problem hiding this comment. Have you considered including this in There was a problem hiding this comment. Never mind - I misunderstood the name of the class and thought it was also uploading the files to the resource staging server as well. |
||
| mainContainerName: String, originalPodSpec: PodBuilder) : PodBuilder | ||
| } | ||
|
|
||
| private[spark] class DriverPodKubernetesFileMounterImpl(filesDownloadPath: String) | ||
| extends DriverPodKubernetesFileMounter { | ||
|
There was a problem hiding this comment. I like the idea of making this generic - we might want to put the submitted jars in here too. It's worth noting that for the next refactor pass. There was a problem hiding this comment. In the above case, I only load file:// into spark-files, but resolve paths and mount for the purpose of the docker image environment variables using this Trait. |
||
| val LocalPattern = "(local://)(.*)".r | ||
|
There was a problem hiding this comment. We've been using |
||
| val FilePattern = "(file:/)(.*)".r | ||
| def getName(file: String, separatorChar: Char) : String = { | ||
| val index: Int = file.lastIndexOf(separatorChar) | ||
| file.substring(index + 1) | ||
| } | ||
| def fileLoc(file: String) : String = file match { | ||
| case "" => "" | ||
| case LocalPattern(_, file_name) => file_name | ||
| case FilePattern(_, file_name) => filesDownloadPath + "/" + getName(file_name, '/') | ||
| case _ => filesDownloadPath + "/" + getName(file, '/') | ||
| } | ||
| def pythonFileLocations(pFiles: List[String], mainAppResource: String) : String = { | ||
|
There was a problem hiding this comment. These operations are mostly shared with ContainerLocalizedFileResolver. Can we use that instead? There was a problem hiding this comment. I have refactored to do this |
||
| def recFileLoc(file: List[String]): List[String] = file match { | ||
| case Nil => List.empty[String] | ||
| case a::b => a match { | ||
| case _ if a==mainAppResource => recFileLoc(b) | ||
| case _ => fileLoc(a) +: recFileLoc(b) | ||
| } | ||
| } | ||
| recFileLoc(pFiles).mkString(",") | ||
| } | ||
| override def addPySparkFiles(mainAppResource: String, pythonFiles: List[String], | ||
| mainContainerName: String, | ||
|
There was a problem hiding this comment. Indentation here - put each argument on its own line. See argument lists for the constructor of |
||
| originalPodSpec: PodBuilder): PodBuilder = { | ||
| originalPodSpec | ||
| .editSpec() | ||
| .editMatchingContainer(new ContainerNameEqualityPredicate(mainContainerName)) | ||
|
There was a problem hiding this comment. We've been indenting these to make it easier to track where the objects begin and end. See for example |
||
| .addNewEnv() | ||
| .withName(ENV_PYSPARK_PRIMARY) | ||
| .withValue(fileLoc(mainAppResource)) | ||
| .endEnv() | ||
| .addNewEnv() | ||
| .withName(ENV_PYSPARK_FILES) | ||
| .withValue(pythonFileLocations(pythonFiles, mainAppResource)) | ||
| .endEnv() | ||
| .endContainer() | ||
| .endSpec() | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| # | ||
| # 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. | ||
| # | ||
|
|
||
| FROM spark-base | ||
|
|
||
| # If this docker file is being used in the context of building your images from a Spark distribution, the docker build | ||
| # command should be invoked from the top level directory of the Spark distribution. E.g.: | ||
| # docker build -t spark-driver-py:latest -f dockerfiles/driver-py/Dockerfile . | ||
|
|
||
| ADD examples /opt/spark/examples | ||
| ADD python /opt/spark/python | ||
|
|
||
| RUN apk add --no-cache python && \ | ||
| python -m ensurepip && \ | ||
| rm -r /usr/lib/python*/ensurepip && \ | ||
| pip install --upgrade pip setuptools && \ | ||
| rm -r /root/.cache | ||
| # UNCOMMENT THE FOLLOWING TO START PIP INSTALLING PYTHON PACKAGES | ||
| # RUN apk add --update alpine-sdk python-dev | ||
| # RUN pip install numpy | ||
|
|
||
| ENV PYTHON_VERSION 2.7.13 | ||
| ENV PYSPARK_PYTHON python | ||
| ENV PYSPARK_DRIVER_PYTHON python | ||
| ENV PYTHONPATH ${SPARK_HOME}/python/:${SPARK_HOME}/python/lib/py4j-0.10.4-src.zip:${PYTHONPATH} | ||
|
|
||
| CMD SPARK_CLASSPATH="${SPARK_HOME}/jars/*" && \ | ||
|
There was a problem hiding this comment. The classpath environment variables may mean different in Pyspark, but given that we might want to also be shipping jars for UDFs then this might still apply. There was a problem hiding this comment. Exactly, I was confused about whether this would be necessary. This is an extension to the question about whether spark.jars should be empty There was a problem hiding this comment. My understanding now is that we actually do need jars for SQL UDFs. cc @robert3005 There was a problem hiding this comment. Okay I will account for that by allowing submission of spark.jars in Client.scala |
||
| if ! [ -z ${SPARK_MOUNTED_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_MOUNTED_CLASSPATH:$SPARK_CLASSPATH"; fi && \ | ||
| if ! [ -z ${SPARK_SUBMIT_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_SUBMIT_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \ | ||
| if ! [ -z ${SPARK_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \ | ||
| if ! [ -z ${SPARK_MOUNTED_FILES_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_DIR/." .; fi && \ | ||
| exec /sbin/tini -- ${JAVA_HOME}/bin/java $SPARK_DRIVER_JAVA_OPTS -cp $SPARK_CLASSPATH \ | ||
| -Xms$SPARK_DRIVER_MEMORY -Xmx$SPARK_DRIVER_MEMORY \ | ||
| $SPARK_DRIVER_CLASS $PYSPARK_PRIMARY $PYSPARK_FILES $SPARK_DRIVER_ARGS | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: could factor out
childArgs += args.primaryResourceThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
val mainAppResource = args(0)so it is necessary to point to the mainAppResource in Client.main()