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 |
|---|---|---|
|
|
@@ -24,13 +24,19 @@ private[spark] trait ContainerLocalizedFilesResolver { | |
| def resolveSubmittedAndRemoteSparkJars(): Seq[String] | ||
| def resolveSubmittedSparkJars(): Seq[String] | ||
| def resolveSubmittedSparkFiles(): Seq[String] | ||
| def resolveSubmittedPySparkFiles(): Seq[String] | ||
| def resolvePrimaryResourceFile(): String | ||
| } | ||
|
|
||
| private[spark] class ContainerLocalizedFilesResolverImpl( | ||
| sparkJars: Seq[String], | ||
| sparkFiles: Seq[String], | ||
| pySparkFiles: Seq[String], | ||
| primaryPyFile: String, | ||
| jarsDownloadPath: String, | ||
| filesDownloadPath: String | ||
| ) extends ContainerLocalizedFilesResolver { | ||
|
There was a problem hiding this comment. Move this line up - notice the diff from what this had before. |
||
|
|
||
|
|
||
| override def resolveSubmittedAndRemoteSparkJars(): Seq[String] = { | ||
| sparkJars.map { jar => | ||
|
|
@@ -53,16 +59,33 @@ private[spark] class ContainerLocalizedFilesResolverImpl( | |
| resolveSubmittedFiles(sparkFiles, filesDownloadPath) | ||
| } | ||
|
|
||
| override def resolveSubmittedPySparkFiles(): Seq[String] = { | ||
| def filterMainResource(x: String) = x match { | ||
| case `primaryPyFile` => None | ||
| case _ => Some(resolveFile(x, filesDownloadPath)) | ||
| } | ||
| pySparkFiles.flatMap(x => filterMainResource(x)) | ||
| } | ||
|
|
||
| override def resolvePrimaryResourceFile(): String = { | ||
| Option(primaryPyFile) match { | ||
|
There was a problem hiding this comment. Use There was a problem hiding this comment. Why not? Is that Spark specific scala practice? There was a problem hiding this comment. Not just spark-specific but seems to be the standard across all of Scala. There was a problem hiding this comment. See https://www.scala-lang.org/api/current/scala/Option.html "The most idiomatic way to use an scala.Option instance is to treat it as a collection or monad and use map,flatMap, filter, or foreach... A less-idiomatic way to use scala.Option values is via pattern matching" |
||
| case None => "" | ||
| case Some(p) => resolveFile(p, filesDownloadPath) | ||
| } | ||
| } | ||
|
|
||
| private def resolveFile(file: String, downloadPath: String) = { | ||
| val fileUri = Utils.resolveURI(file) | ||
| Option(fileUri.getScheme).getOrElse("file") match { | ||
| case "file" => | ||
| val fileName = new File(fileUri.getPath).getName | ||
| s"$downloadPath/$fileName" | ||
| case _ => | ||
| file | ||
| } | ||
| } | ||
|
|
||
| private def resolveSubmittedFiles(files: Seq[String], downloadPath: String): Seq[String] = { | ||
| files.map { file => resolveFile(file, downloadPath) } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /* | ||
| * 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(primaryFile: String, pySparkFiles: String, | ||
| mainContainerName: String, originalPodSpec: PodBuilder) : PodBuilder | ||
| } | ||
|
|
||
| private[spark] class DriverPodKubernetesFileMounterImpl() | ||
| 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. |
||
| override def addPySparkFiles( | ||
| primaryFile: String, | ||
| pySparkFiles: String, | ||
| mainContainerName: String, | ||
| originalPodSpec: PodBuilder): PodBuilder = { | ||
|
|
||
| originalPodSpec | ||
| .editSpec() | ||
| .editMatchingContainer(new ContainerNameEqualityPredicate(mainContainerName)) | ||
| .addNewEnv() | ||
| .withName(ENV_PYSPARK_PRIMARY) | ||
| .withValue(primaryFile) | ||
| .endEnv() | ||
| .addNewEnv() | ||
| .withName(ENV_PYSPARK_FILES) | ||
| .withValue(pySparkFiles) | ||
| .endEnv() | ||
| .endContainer() | ||
| .endSpec() | ||
| } | ||
| } | ||
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()