You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by mc...@apache.org on 2018/09/06 23:19:04 UTC
spark git commit: [SPARK-25262][K8S] Allow SPARK_LOCAL_DIRS to be
tmpfs backed on K8S
Repository: spark
Updated Branches:
refs/heads/master 27d3b0a51 -> da6fa3828
[SPARK-25262][K8S] Allow SPARK_LOCAL_DIRS to be tmpfs backed on K8S
## What changes were proposed in this pull request?
The default behaviour of Spark on K8S currently is to create `emptyDir` volumes to back `SPARK_LOCAL_DIRS`. In some environments e.g. diskless compute nodes this may actually hurt performance because these are backed by the Kubelet's node storage which on a diskless node will typically be some remote network storage.
Even if this is enterprise grade storage connected via a high speed interconnect the way Spark uses these directories as scratch space (lots of relatively small short lived files) has been observed to cause serious performance degradation. Therefore we would like to provide the option to use K8S's ability to instead back these `emptyDir` volumes with `tmpfs`. Therefore this PR adds a configuration option that enables `SPARK_LOCAL_DIRS` to be backed by Memory backed `emptyDir` volumes rather than the default.
Documentation is added to describe both the default behaviour plus this new option and its implications. One of which is that scratch space then counts towards your pods memory limits and therefore users will need to adjust their memory requests accordingly.
*NB* - This is an alternative version of PR #22256 reduced to just the `tmpfs` piece
## How was this patch tested?
Ran with this option in our diskless compute environments to verify functionality
Author: Rob Vesse <rv...@dotnetrdf.org>
Closes #22323 from rvesse/SPARK-25262-tmpfs.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/da6fa382
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/da6fa382
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/da6fa382
Branch: refs/heads/master
Commit: da6fa3828bb824b65f50122a8a0a0d4741551257
Parents: 27d3b0a
Author: Rob Vesse <rv...@dotnetrdf.org>
Authored: Thu Sep 6 16:18:59 2018 -0700
Committer: mcheah <mc...@palantir.com>
Committed: Thu Sep 6 16:18:59 2018 -0700
----------------------------------------------------------------------
docs/running-on-kubernetes.md | 21 ++++++++++++++
.../org/apache/spark/deploy/k8s/Config.scala | 9 ++++++
.../k8s/features/LocalDirsFeatureStep.scala | 3 ++
.../features/LocalDirsFeatureStepSuite.scala | 30 ++++++++++++++++++++
4 files changed, 63 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/da6fa382/docs/running-on-kubernetes.md
----------------------------------------------------------------------
diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index c83dad6..4ae7aca 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -215,6 +215,19 @@ spark.kubernetes.driver.volumes.persistentVolumeClaim.checkpointpvc.options.clai
The configuration properties for mounting volumes into the executor pods use prefix `spark.kubernetes.executor.` instead of `spark.kubernetes.driver.`. For a complete list of available options for each supported type of volumes, please refer to the [Spark Properties](#spark-properties) section below.
+## Local Storage
+
+Spark uses temporary scratch space to spill data to disk during shuffles and other operations. When using Kubernetes as the resource manager the pods will be created with an [emptyDir](https://kubernetes.io/docs/concepts/storage/volumes/#emptydir) volume mounted for each directory listed in `SPARK_LOCAL_DIRS`. If no directories are explicitly specified then a default directory is created and configured appropriately.
+
+`emptyDir` volumes use the ephemeral storage feature of Kubernetes and do not persist beyond the life of the pod.
+
+### Using RAM for local storage
+
+`emptyDir` volumes use the nodes backing storage for ephemeral storage by default, this behaviour may not be appropriate for some compute environments. For example if you have diskless nodes with remote storage mounted over a network, having lots of executors doing IO to this remote storage may actually degrade performance.
+
+In this case it may be desirable to set `spark.kubernetes.local.dirs.tmpfs=true` in your configuration which will cause the `emptyDir` volumes to be configured as `tmpfs` i.e. RAM backed volumes. When configured like this Sparks local storage usage will count towards your pods memory usage therefore you may wish to increase your memory requests by increasing the value of `spark.kubernetes.memoryOverheadFactor` as appropriate.
+
+
## Introspection and Debugging
These are the different ways in which you can investigate a running/completed Spark application, monitor progress, and
@@ -785,6 +798,14 @@ specific to Spark on Kubernetes.
</td>
</tr>
<tr>
+ <td><code>spark.kubernetes.local.dirs.tmpfs</code></td>
+ <td><code>false</code>
+ <td>
+ Configure the <code>emptyDir</code> volumes used to back <code>SPARK_LOCAL_DIRS</code> within the Spark driver and executor pods to use <code>tmpfs</code> backing i.e. RAM. See <a href="#local-storage">Local Storage</a> earlier on this page
+ for more discussion of this.
+ </td>
+</tr>
+<tr>
<td><code>spark.kubernetes.memoryOverheadFactor</code></td>
<td><code>0.1</code></td>
<td>
http://git-wip-us.apache.org/repos/asf/spark/blob/da6fa382/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
----------------------------------------------------------------------
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
index 1b582fe..c5f4d6c 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -225,6 +225,15 @@ private[spark] object Config extends Logging {
"Ensure that major Python version is either Python2 or Python3")
.createWithDefault("2")
+ val KUBERNETES_LOCAL_DIRS_TMPFS =
+ ConfigBuilder("spark.kubernetes.local.dirs.tmpfs")
+ .doc("If set to true then emptyDir volumes created to back SPARK_LOCAL_DIRS will have " +
+ "their medium set to Memory so that they will be created as tmpfs (i.e. RAM) backed " +
+ "volumes. This may improve performance but scratch space usage will count towards " +
+ "your pods memory limit so you may wish to request more memory.")
+ .booleanConf
+ .createWithDefault(false)
+
val KUBERNETES_AUTH_SUBMISSION_CONF_PREFIX =
"spark.kubernetes.authenticate.submission"
http://git-wip-us.apache.org/repos/asf/spark/blob/da6fa382/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala
----------------------------------------------------------------------
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala
index 70b3073..be386e1 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala
@@ -22,6 +22,7 @@ import java.util.UUID
import io.fabric8.kubernetes.api.model.{ContainerBuilder, HasMetadata, PodBuilder, VolumeBuilder, VolumeMountBuilder}
import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverSpecificConf, KubernetesRoleSpecificConf, SparkPod}
+import org.apache.spark.deploy.k8s.Config._
private[spark] class LocalDirsFeatureStep(
conf: KubernetesConf[_ <: KubernetesRoleSpecificConf],
@@ -37,6 +38,7 @@ private[spark] class LocalDirsFeatureStep(
.orElse(conf.getOption("spark.local.dir"))
.getOrElse(defaultLocalDir)
.split(",")
+ private val useLocalDirTmpFs = conf.get(KUBERNETES_LOCAL_DIRS_TMPFS)
override def configurePod(pod: SparkPod): SparkPod = {
val localDirVolumes = resolvedLocalDirs
@@ -45,6 +47,7 @@ private[spark] class LocalDirsFeatureStep(
new VolumeBuilder()
.withName(s"spark-local-dir-${index + 1}")
.withNewEmptyDir()
+ .withMedium(if (useLocalDirTmpFs) "Memory" else null)
.endEmptyDir()
.build()
}
http://git-wip-us.apache.org/repos/asf/spark/blob/da6fa382/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala
----------------------------------------------------------------------
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala
index a339827..acdd07b 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala
@@ -18,10 +18,12 @@ package org.apache.spark.deploy.k8s.features
import io.fabric8.kubernetes.api.model.{EnvVarBuilder, VolumeBuilder, VolumeMountBuilder}
import org.mockito.Mockito
+import org.scalatest._
import org.scalatest.BeforeAndAfter
import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverSpecificConf, KubernetesRoleSpecificConf, SparkPod}
+import org.apache.spark.deploy.k8s.Config._
class LocalDirsFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
private val defaultLocalDir = "/var/data/default-local-dir"
@@ -111,4 +113,32 @@ class LocalDirsFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
.withValue("/var/data/my-local-dir-1,/var/data/my-local-dir-2")
.build())
}
+
+ test("Use tmpfs to back default local dir") {
+ Mockito.doReturn(null).when(sparkConf).get("spark.local.dir")
+ Mockito.doReturn(null).when(sparkConf).getenv("SPARK_LOCAL_DIRS")
+ Mockito.doReturn(true).when(sparkConf).get(KUBERNETES_LOCAL_DIRS_TMPFS)
+ val stepUnderTest = new LocalDirsFeatureStep(kubernetesConf, defaultLocalDir)
+ val configuredPod = stepUnderTest.configurePod(SparkPod.initialPod())
+ assert(configuredPod.pod.getSpec.getVolumes.size === 1)
+ assert(configuredPod.pod.getSpec.getVolumes.get(0) ===
+ new VolumeBuilder()
+ .withName(s"spark-local-dir-1")
+ .withNewEmptyDir()
+ .withMedium("Memory")
+ .endEmptyDir()
+ .build())
+ assert(configuredPod.container.getVolumeMounts.size === 1)
+ assert(configuredPod.container.getVolumeMounts.get(0) ===
+ new VolumeMountBuilder()
+ .withName(s"spark-local-dir-1")
+ .withMountPath(defaultLocalDir)
+ .build())
+ assert(configuredPod.container.getEnv.size === 1)
+ assert(configuredPod.container.getEnv.get(0) ===
+ new EnvVarBuilder()
+ .withName("SPARK_LOCAL_DIRS")
+ .withValue(defaultLocalDir)
+ .build())
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org