diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index 975b28de47e2..9c4644947c91 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -549,14 +549,23 @@ specific to Spark on Kubernetes.
spark.kubernetes.driver.limit.cores |
(none) |
- Specify the hard CPU [limit](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#resource-requests-and-limits-of-pod-and-container) for the driver pod.
+ Specify a hard cpu [limit](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#resource-requests-and-limits-of-pod-and-container) for the driver pod.
|
+
+ spark.kubernetes.executor.request.cores |
+ (none) |
+
+ Specify the cpu request for each executor pod. Values conform to the Kubernetes [convention](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#meaning-of-cpu).
+ Example values include 0.1, 500m, 1.5, 5, etc., with the definition of cpu units documented in [CPU units](https://kubernetes.io/docs/tasks/configure-pod-container/assign-cpu-resource/#cpu-units).
+ This is distinct from spark.executor.cores: it is only used and takes precedence over spark.executor.cores for specifying the executor pod cpu request if set. Task
+ parallelism, e.g., number of tasks an executor can run concurrently is not affected by this.
+ |
spark.kubernetes.executor.limit.cores |
(none) |
- Specify the hard CPU [limit](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#resource-requests-and-limits-of-pod-and-container) for each executor pod launched for the Spark Application.
+ Specify a hard cpu [limit](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#resource-requests-and-limits-of-pod-and-container) for each executor pod launched for the Spark Application.
|
@@ -593,4 +602,4 @@ specific to Spark on Kubernetes.
spark.kubernetes.executor.secrets.spark-secret=/etc/secrets.
-
\ No newline at end of file
+
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 da34a7e06238..405ea476351b 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
@@ -91,6 +91,12 @@ private[spark] object Config extends Logging {
.stringConf
.createOptional
+ val KUBERNETES_EXECUTOR_REQUEST_CORES =
+ ConfigBuilder("spark.kubernetes.executor.request.cores")
+ .doc("Specify the cpu request for each executor pod")
+ .stringConf
+ .createOptional
+
val KUBERNETES_DRIVER_POD_NAME =
ConfigBuilder("spark.kubernetes.driver.pod.name")
.doc("Name of the driver pod.")
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala
index 98cbd5607da0..44d4d4d243f7 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala
@@ -83,7 +83,12 @@ private[spark] class ExecutorPodFactory(
MEMORY_OVERHEAD_MIN_MIB))
private val executorMemoryWithOverhead = executorMemoryMiB + memoryOverheadMiB
- private val executorCores = sparkConf.getDouble("spark.executor.cores", 1)
+ private val executorCores = sparkConf.getInt("spark.executor.cores", 1)
+ private val executorCoresRequest = if (sparkConf.contains(KUBERNETES_EXECUTOR_REQUEST_CORES)) {
+ sparkConf.get(KUBERNETES_EXECUTOR_REQUEST_CORES).get
+ } else {
+ executorCores.toString
+ }
private val executorLimitCores = sparkConf.get(KUBERNETES_EXECUTOR_LIMIT_CORES)
/**
@@ -114,7 +119,7 @@ private[spark] class ExecutorPodFactory(
.withAmount(s"${executorMemoryWithOverhead}Mi")
.build()
val executorCpuQuantity = new QuantityBuilder(false)
- .withAmount(executorCores.toString)
+ .withAmount(executorCoresRequest)
.build()
val executorExtraClasspathEnv = executorExtraClasspath.map { cp =>
new EnvVarBuilder()
@@ -133,8 +138,7 @@ private[spark] class ExecutorPodFactory(
}.getOrElse(Seq.empty[EnvVar])
val executorEnv = (Seq(
(ENV_DRIVER_URL, driverUrl),
- // Executor backend expects integral value for executor cores, so round it up to an int.
- (ENV_EXECUTOR_CORES, math.ceil(executorCores).toInt.toString),
+ (ENV_EXECUTOR_CORES, executorCores.toString),
(ENV_EXECUTOR_MEMORY, executorMemoryString),
(ENV_APPLICATION_ID, applicationId),
// This is to set the SPARK_CONF_DIR to be /opt/spark/conf
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactorySuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactorySuite.scala
index 7755b9383504..4922baff51dd 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactorySuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactorySuite.scala
@@ -83,6 +83,33 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
checkOwnerReferences(executor, driverPodUid)
}
+ test("executor core request specification") {
+ var factory = new ExecutorPodFactory(baseConf, None)
+ var executor = factory.createExecutorPod(
+ "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
+ assert(executor.getSpec.getContainers.size() === 1)
+ assert(executor.getSpec.getContainers.get(0).getResources.getRequests.get("cpu").getAmount
+ === "1")
+
+ val conf = baseConf.clone()
+
+ conf.set(KUBERNETES_EXECUTOR_REQUEST_CORES, "0.1")
+ factory = new ExecutorPodFactory(conf, None)
+ executor = factory.createExecutorPod(
+ "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
+ assert(executor.getSpec.getContainers.size() === 1)
+ assert(executor.getSpec.getContainers.get(0).getResources.getRequests.get("cpu").getAmount
+ === "0.1")
+
+ conf.set(KUBERNETES_EXECUTOR_REQUEST_CORES, "100m")
+ factory = new ExecutorPodFactory(conf, None)
+ conf.set(KUBERNETES_EXECUTOR_REQUEST_CORES, "100m")
+ executor = factory.createExecutorPod(
+ "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
+ assert(executor.getSpec.getContainers.get(0).getResources.getRequests.get("cpu").getAmount
+ === "100m")
+ }
+
test("executor pod hostnames get truncated to 63 characters") {
val conf = baseConf.clone()
conf.set(KUBERNETES_EXECUTOR_POD_NAME_PREFIX,