This repository was archived by the owner on Jan 9, 2020. It is now read-only.
forked from apache/spark
-
Notifications
You must be signed in to change notification settings - Fork 117
Mount emptyDir volumes for temporary directories on executors in static allocation mode (rebased) #522
Merged
ash211
merged 8 commits into
branch-2.2-kubernetes
from
use-empty-dir-static-allocation-shuffles-rebased
Oct 16, 2017
Merged
Mount emptyDir volumes for temporary directories on executors in static allocation mode (rebased) #522
Changes from all commits
Commits
Show all changes
8 commits
Select commit
Hold shift + click to select a range
eb017f2
Use emptyDir volume mounts for executor local directories.
mccheah ba3c486
Mount local dirs in the driver. Remove shuffle dir configuration.
mccheah a6ef915
Arrange imports
mccheah f2471a6
Fix style and integration tests.
mccheah fb414c3
Add TODO note for volume types to change.
mccheah 879207a
Add unit test and extra documentation.
mccheah 02b283e
Fix existing unit tests and add tests for empty dir volumes
mccheah 463cf3f
Remove extraneous constant
mccheah File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
98 changes: 98 additions & 0 deletions
98
...org/apache/spark/deploy/k8s/submit/submitsteps/LocalDirectoryMountConfigurationStep.scala
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,98 @@ | ||
| /* | ||
| * 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.k8s.submit.submitsteps | ||
|
|
||
| import java.nio.file.Paths | ||
| import java.util.UUID | ||
|
|
||
| import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder, VolumeBuilder, VolumeMountBuilder} | ||
|
|
||
| import org.apache.spark.SparkConf | ||
| import org.apache.spark.deploy.k8s.constants._ | ||
|
|
||
| /** | ||
| * Configures local directories that the driver and executors should use for temporary storage. | ||
| * | ||
| * Note that we have different semantics for scratch space in Kubernetes versus the other cluster | ||
| * managers. In Kubernetes, we cannot allow the local directories to resolve to the Java temporary | ||
| * directory. This is because we will mount either emptyDir volumes for both the driver and | ||
| * executors, or hostPath volumes for the executors and an emptyDir for the driver. In either | ||
| * case, the mount paths need to be directories that do not exist in the base container images. | ||
| * But the Java temporary directory is typically a directory like /tmp which exists in most | ||
| * container images. | ||
| * | ||
| * The solution is twofold: | ||
| * - When not using an external shuffle service, a reasonable default is to create a new directory | ||
| * with a random name and set that to be the value of `spark.local.dir`. | ||
| * - When using the external shuffle service, it is risky to assume that the user intends to mount | ||
| * the JVM temporary directory into the pod as a hostPath volume. We therefore enforce that | ||
| * spark.local.dir must be set in dynamic allocation mode so that the user explicitly sets the | ||
| * paths that have to be mounted. | ||
| */ | ||
| private[spark] class LocalDirectoryMountConfigurationStep( | ||
| submissionSparkConf: SparkConf, | ||
| randomDirProvider: () => String = () => s"spark-${UUID.randomUUID()}") | ||
| extends DriverConfigurationStep { | ||
|
|
||
| override def configureDriver(driverSpec: KubernetesDriverSpec): KubernetesDriverSpec = { | ||
| val configuredLocalDirs = submissionSparkConf.getOption("spark.local.dir") | ||
| val isUsingExternalShuffle = submissionSparkConf.get( | ||
| org.apache.spark.internal.config.SHUFFLE_SERVICE_ENABLED) | ||
| val resolvedLocalDirsSingleString = if (isUsingExternalShuffle) { | ||
| require(configuredLocalDirs.isDefined, "spark.local.dir must be provided explicitly when" + | ||
| " using the external shuffle service in Kubernetes. These directories should map to" + | ||
| " the paths that are mounted into the external shuffle service pods.") | ||
| configuredLocalDirs.get | ||
| } else { | ||
| // If we don't use the external shuffle service, local directories should be randomized if | ||
| // not provided. | ||
| configuredLocalDirs.getOrElse(s"$GENERATED_LOCAL_DIR_MOUNT_ROOT/${randomDirProvider()}") | ||
| } | ||
| val resolvedLocalDirs = resolvedLocalDirsSingleString.split(",") | ||
| // It's worth noting that we always use an emptyDir volume for the directories on the driver, | ||
| // because the driver does not need a hostPath to share its scratch space with any other pod. | ||
| // The driver itself will decide on whether to use a hostPath volume or an emptyDir volume for | ||
| // these directories on the executors. (see ExecutorPodFactory and | ||
| // KubernetesExternalClusterManager) | ||
| val localDirVolumes = resolvedLocalDirs.zipWithIndex.map { case (dir, index) => | ||
| new VolumeBuilder() | ||
| .withName(s"spark-local-dir-$index-${Paths.get(dir).getFileName.toString}") | ||
| .withNewEmptyDir().endEmptyDir() | ||
| .build() | ||
| } | ||
| val localDirVolumeMounts = localDirVolumes.zip(resolvedLocalDirs).map { | ||
| case (volume, path) => | ||
| new VolumeMountBuilder() | ||
| .withName(volume.getName) | ||
| .withMountPath(path) | ||
| .build() | ||
| } | ||
| val resolvedDriverSparkConf = driverSpec.driverSparkConf.clone().set( | ||
| "spark.local.dir", resolvedLocalDirsSingleString) | ||
| driverSpec.copy( | ||
| driverPod = new PodBuilder(driverSpec.driverPod) | ||
| .editSpec() | ||
| .addToVolumes(localDirVolumes: _*) | ||
| .endSpec() | ||
| .build(), | ||
| driverContainer = new ContainerBuilder(driverSpec.driverContainer) | ||
| .addToVolumeMounts(localDirVolumeMounts: _*) | ||
| .build(), | ||
| driverSparkConf = resolvedDriverSparkConf | ||
| ) | ||
| } | ||
| } |
61 changes: 61 additions & 0 deletions
61
...rc/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorLocalDirVolumeProvider.scala
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| /* | ||
| * 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.scheduler.cluster.k8s | ||
|
|
||
| import java.nio.file.Paths | ||
|
|
||
| import io.fabric8.kubernetes.api.model.{Volume, VolumeBuilder, VolumeMount, VolumeMountBuilder} | ||
|
|
||
| import org.apache.spark.SparkConf | ||
| import org.apache.spark.util.Utils | ||
|
|
||
| private[spark] trait ExecutorLocalDirVolumeProvider { | ||
| def getExecutorLocalDirVolumesWithMounts: Seq[(Volume, VolumeMount)] | ||
| } | ||
|
|
||
| private[spark] class ExecutorLocalDirVolumeProviderImpl( | ||
| sparkConf: SparkConf, | ||
| kubernetesExternalShuffleManager: Option[KubernetesExternalShuffleManager]) | ||
| extends ExecutorLocalDirVolumeProvider { | ||
| override def getExecutorLocalDirVolumesWithMounts: Seq[(Volume, VolumeMount)] = { | ||
| kubernetesExternalShuffleManager.map(_.getExecutorShuffleDirVolumesWithMounts) | ||
| .getOrElse { | ||
| // If we're not using the external shuffle manager, we should use emptyDir volumes for | ||
| // shuffle directories since it's important for disk I/O for these directories to be | ||
| // performant. If the user has not provided a local directory, instead of using the | ||
| // Java temporary directory, we create one instead, because we want to avoid | ||
| // mounting an emptyDir which overlaps with an existing path in the Docker image. | ||
| // Java's temporary directory path is typically /tmp or a similar path, which is | ||
| // likely to exist in most images. | ||
| val resolvedLocalDirs = Utils.getConfiguredLocalDirs(sparkConf) | ||
| val localDirVolumes = resolvedLocalDirs.zipWithIndex.map { case (dir, index) => | ||
| new VolumeBuilder() | ||
| .withName(s"spark-local-dir-$index-${Paths.get(dir).getFileName.toString}") | ||
| .withNewEmptyDir().endEmptyDir() | ||
| .build() | ||
| } | ||
| val localDirVolumeMounts = localDirVolumes.zip(resolvedLocalDirs).map { | ||
| case (volume, path) => | ||
| new VolumeMountBuilder() | ||
| .withName(volume.getName) | ||
| .withMountPath(path) | ||
| .build() | ||
| } | ||
| localDirVolumes.zip(localDirVolumeMounts) | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
was trying to think through if there are any security concerns here, but I don't think there are any additional ones. Any data in the broadcast/cache files would with high probability also be accessible in shuffle files, so there's no new data exposed. We have other methods to secure the shuffle dirs between peer executors on the same node