You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "DenineLu (via GitHub)" <gi...@apache.org> on 2026/03/09 12:44:50 UTC
[PR] [SPARK-55888] [K8S] Parallel pod creation for Kubernetes executor allocation [spark]
DenineLu opened a new pull request, #54688:
URL: https://github.com/apache/spark/pull/54688
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
This PR adds support for creating executor pods in parallel during Kubernetes executor allocation.
Right now, `ExecutorPodsAllocator` creates pods one by one — it sends a create request to the K8s API, waits for the response, then moves on to the next one. This works fine for small jobs, but becomes a real bottleneck when you need thousands of executors.
The idea here is straightforward: pod spec building still happens serially (since it touches shared mutable state like EXECUTOR_ID_COUNTER and reusablePVCs), but the actual K8s API calls to create pods are fired off concurrently using a fixed-size thread pool. After all the futures complete, successful results are registered back on the caller thread.
A new config `spark.kubernetes.allocation.batch.concurrency` controls the parallelism level. It defaults to 1, which keeps the original serial behavior — so this is fully backward-compatible.
### Why are the changes needed?
Each pod creation involves a round-trip to the K8s API server, and depending on cluster load, admission webhooks, etcd, etc., each call can take tens to hundreds of milliseconds. When you're creating 10,000 executors, that adds up fast.
I ran some benchmarks with batchSize=100, maxPending=1000, allocating 10,000 executors.
In production environments, K8s clusters typically have various admission webhooks enabled (security policies, resource quotas, sidecar injection, etc.), and some of them can add significant latency to pod creation calls. In our case, certain webhooks were responding slowly, and the serial allocation became a serious pain point:
| Concurrency | Time (s) | Speedup |
|--------|--------|--------|
| 1 (serial) | 3412 | - |
| 5 | 1171 | 2.91x |
| 10 | 1269 | 2.69x |
| 20 | 1309 | 2.61x |
Serial allocation took nearly 57 minutes to spin up 10,000 executors. With concurrency=5, that dropped to ~20 minutes — a ~66% reduction.
After disabling those high-latency webhooks, the per-call latency went down, but even then, production K8s clusters still have inherent overhead (API server processing, etcd writes, other controllers reacting to pod events, etc.) that adds up at scale:
| Concurrency | Time (s) | Speedup |
|--------|--------|--------|
| 1 (serial) | 755 | - |
| 3 | 545 | 1.39x |
| 5 | 522 | 1.45x |
| 10 | 588 | 1.28x |
Even in this lower-latency scenario, concurrency=5 still shaved off ~30% of allocation time. Setting concurrency too high (like 10) can actually hurt a bit, likely due to increased contention on the K8s API server — so a moderate value like 3–5 seems to be the sweet spot.
### Does this PR introduce _any_ user-facing change?
Yes. A new config is added:
- `spark.kubernetes.allocation.batch.concurrency` (default: 1): how many threads to use for creating executor pods concurrently. Set to 1 for the original serial behavior.
### How was this patch tested?
All existing tests in ExecutorPodsAllocatorSuite pass (they run with concurrency=1).
Added a new test for the parallel path: verifies that PVC creation failures don't leak the PVC counter.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "DenineLu (via GitHub)" <gi...@apache.org>.
DenineLu commented on code in PR #54688:
URL: https://github.com/apache/spark/pull/54688#discussion_r2929511627
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -574,6 +711,11 @@ class ExecutorPodsAllocator(
}
override def stop(applicationId: String): Unit = {
+ if (podAllocationConcurrency > 1) {
+ Utils.tryLogNonFatalError {
+ podCreationThreadPool.shutdownNow()
Review Comment:
> Even if `snapshotsStore.stop()` is invoked first, it does not guarantee thread `subscribersExecutor` stops first.
>
> A thread pool's shutdown does not terminate threads that do not respond to interrupts.
You're right, I caught the exception and performed the necessary checks before committing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "yikf (via GitHub)" <gi...@apache.org>.
yikf commented on code in PR #54688:
URL: https://github.com/apache/spark/pull/54688#discussion_r2922104197
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -574,6 +711,11 @@ class ExecutorPodsAllocator(
}
override def stop(applicationId: String): Unit = {
+ if (podAllocationConcurrency > 1) {
+ Utils.tryLogNonFatalError {
+ podCreationThreadPool.shutdownNow()
Review Comment:
If `onNewSnapshots` is triggered by [the scheduler thread](https://github.com/apache/spark/blob/8d645022912715201da122b654155b9f1edb9316/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala#L84), what happens if we stop the thread pool while the scheduler thread is still running?
Will it throw a RejectedExecutionException? Although it may not matter, since Spark is in the process of stopping.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -438,6 +448,116 @@ class ExecutorPodsAllocator(
applicationId: String,
resourceProfileId: Int,
pvcsInUse: Seq[String]): Unit = {
+ if (podAllocationConcurrency > 1) {
Review Comment:
It would be better to add a enable configuration.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -446,77 +566,94 @@ class ExecutorPodsAllocator(
log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT, PVC_COUNTER.get())} PVCs.")
return
}
- val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
- if (newExecutorId >= podAllocationMaximum) {
- throw new SparkException(s"Exceed the pod creation limit: $podAllocationMaximum")
+ val (newExecutorId, podWithAttachedContainer, resources) =
+ buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ createExecutorPodAndPVCs(newExecutorId, podWithAttachedContainer, resources).foreach { _ =>
+ newlyCreatedExecutors(newExecutorId) = (resourceProfileId, clock.getTimeMillis())
}
- val executorConf = KubernetesConf.createExecutorConf(
- conf,
- newExecutorId.toString,
- applicationId,
- driverPod,
- resourceProfileId)
- val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf, secMgr,
- kubernetesClient, rpIdToResourceProfile(resourceProfileId))
- val executorPod = resolvedExecutorSpec.pod
- val podWithAttachedContainer = new PodBuilder(executorPod.pod)
- .editOrNewSpec()
- .addToContainers(executorPod.container)
- .endSpec()
- .build()
- val resources = replacePVCsIfNeeded(
- podWithAttachedContainer, resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
- val optCreatedExecutorPod = try {
- Some(kubernetesClient
- .pods()
- .inNamespace(namespace)
- .resource(podWithAttachedContainer)
- .create())
- } catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to create executor pod ${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- None
+ }
+ }
+
+ /**
+ * Parallel pod creation: builds pod specs serially (they depend on shared mutable state like
+ * EXECUTOR_ID_COUNTER and reusablePVCs), then creates them via K8s API in parallel using a
+ * thread pool sized by `spark.kubernetes.allocation.batch.concurrency`.
+ */
+ private def requestNewExecutorsParallel(
+ numExecutorsToAllocate: Int,
+ applicationId: String,
+ resourceProfileId: Int,
+ pvcsInUse: Seq[String]): Unit = {
+ val batchStartTime = clock.getTimeMillis()
+
+ // Check reusable PVCs for this executor allocation batch
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+
+ // Phase 1: Build all pod specs serially
+ val podSpecs = mutable.ArrayBuffer.empty[(Long, Pod, Seq[HasMetadata])]
+ var i = 0
+ while (i < numExecutorsToAllocate) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <= PVC_COUNTER.get()) {
+ logInfo(
+ log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT, PVC_COUNTER.get())} PVCs.")
+ // Stop building more specs; proceed to create whatever we've built so far
+ i = numExecutorsToAllocate
+ } else {
+ podSpecs += buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ i += 1
}
- optCreatedExecutorPod.foreach { createdExecutorPod =>
+ }
+ if (podSpecs.isEmpty) return
+
+ val podBuildTotalMs = clock.getTimeMillis() - batchStartTime
+
+ // Phase 2: Create pods via K8s API in parallel
+ val k8sApiStartTime = clock.getTimeMillis()
+
+ implicit val ec: ExecutionContext = podCreationExecutionContext
+
+ val futures = podSpecs.map { case (newExecutorId, podWithAttachedContainer, resources) =>
+ Future {
try {
- addOwnerReference(createdExecutorPod, resources)
- resources
- .filter(_.getKind == "PersistentVolumeClaim")
- .foreach { resource =>
- if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
- addOwnerReference(driverPod.get, Seq(resource))
- }
- val pvc = resource.asInstanceOf[PersistentVolumeClaim]
- logInfo(log"Trying to create PersistentVolumeClaim " +
- log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)} with " +
- log"StorageClass ${MDC(LogKeys.CLASS_NAME, pvc.getSpec.getStorageClassName)}")
- kubernetesClient
- .persistentVolumeClaims()
- .inNamespace(namespace)
- .resource(pvc)
- .create()
- PVC_COUNTER.incrementAndGet()
- }
- newlyCreatedExecutors(newExecutorId) = (resourceProfileId, clock.getTimeMillis())
- logDebug(s"Requested executor with id $newExecutorId from Kubernetes.")
+ val result = createExecutorPodAndPVCs(newExecutorId, podWithAttachedContainer, resources)
+ result.map(execId => (execId, resourceProfileId, clock.getTimeMillis()))
} catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to add owner reference or create PVC for executor pod " +
- log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- kubernetesClient.pods()
- .inNamespace(namespace)
- .resource(createdExecutorPod)
- .delete()
- throw e
+ // createExecutorPodAndPVCs already logs the error and registers the failure
+ // before rethrowing. We catch here only to prevent a failed Future from
+ // short-circuiting Future.sequence, so that other concurrent pods can proceed.
+ case NonFatal(_) => None
}
}
}
+
+ val results = try {
+ val allResults = ThreadUtils.awaitResult(
+ Future.sequence(futures), Duration(podCreationTimeout, TimeUnit.MILLISECONDS))
+ allResults.flatten
+ } catch {
+ case NonFatal(e) =>
+ logError(log"Timed out waiting for parallel pod creation to complete " +
+ log"after ${MDC(LogKeys.TIMEOUT, podCreationTimeout)} ms", e)
+ Seq.empty
Review Comment:
If one creation action times out, none of the Pods in this batch will be considered successfully created. this should **not** be batch-atomic, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "DenineLu (via GitHub)" <gi...@apache.org>.
DenineLu commented on code in PR #54688:
URL: https://github.com/apache/spark/pull/54688#discussion_r2922973245
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -574,6 +711,11 @@ class ExecutorPodsAllocator(
}
override def stop(applicationId: String): Unit = {
+ if (podAllocationConcurrency > 1) {
+ Utils.tryLogNonFatalError {
+ podCreationThreadPool.shutdownNow()
Review Comment:
Good question! This actually won't happen because of the shutdown ordering in KubernetesClusterSchedulerBackend.stop():
```scala
override def stop(): Unit = {
// ...
snapshotsStore.stop() // ← (1) stops the subscriber executor first
// ...
podAllocator.stop(...) // ← (2) shuts down the pod creation thread pool
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "DenineLu (via GitHub)" <gi...@apache.org>.
DenineLu commented on code in PR #54688:
URL: https://github.com/apache/spark/pull/54688#discussion_r2923028289
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -438,6 +448,116 @@ class ExecutorPodsAllocator(
applicationId: String,
resourceProfileId: Int,
pvcsInUse: Seq[String]): Unit = {
+ if (podAllocationConcurrency > 1) {
Review Comment:
> It would be better to add a enable configuration.
Added spark.kubernetes.allocation.batch.parallel.enabled (default false) as an explicit switch. spark.kubernetes.allocation.batch.concurrency now only takes effect when the parallel mode is enabled, and its default is changed to 5.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #54688:
URL: https://github.com/apache/spark/pull/54688#issuecomment-4940829837
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888] [K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "DenineLu (via GitHub)" <gi...@apache.org>.
DenineLu commented on PR #54688:
URL: https://github.com/apache/spark/pull/54688#issuecomment-4036767479
@dongjoon-hyun @HyukjinKwon Could you please review this PR when you have time? Many Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #54688: [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation
URL: https://github.com/apache/spark/pull/54688
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "DenineLu (via GitHub)" <gi...@apache.org>.
DenineLu commented on PR #54688:
URL: https://github.com/apache/spark/pull/54688#issuecomment-4168348222
Hi @dongjoon-hyun, just a gentle ping on this PR when you have a moment. Would really appreciate your feedback, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "DenineLu (via GitHub)" <gi...@apache.org>.
DenineLu commented on code in PR #54688:
URL: https://github.com/apache/spark/pull/54688#discussion_r2922994969
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -446,77 +566,94 @@ class ExecutorPodsAllocator(
log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT, PVC_COUNTER.get())} PVCs.")
return
}
- val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
- if (newExecutorId >= podAllocationMaximum) {
- throw new SparkException(s"Exceed the pod creation limit: $podAllocationMaximum")
+ val (newExecutorId, podWithAttachedContainer, resources) =
+ buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ createExecutorPodAndPVCs(newExecutorId, podWithAttachedContainer, resources).foreach { _ =>
+ newlyCreatedExecutors(newExecutorId) = (resourceProfileId, clock.getTimeMillis())
}
- val executorConf = KubernetesConf.createExecutorConf(
- conf,
- newExecutorId.toString,
- applicationId,
- driverPod,
- resourceProfileId)
- val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf, secMgr,
- kubernetesClient, rpIdToResourceProfile(resourceProfileId))
- val executorPod = resolvedExecutorSpec.pod
- val podWithAttachedContainer = new PodBuilder(executorPod.pod)
- .editOrNewSpec()
- .addToContainers(executorPod.container)
- .endSpec()
- .build()
- val resources = replacePVCsIfNeeded(
- podWithAttachedContainer, resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
- val optCreatedExecutorPod = try {
- Some(kubernetesClient
- .pods()
- .inNamespace(namespace)
- .resource(podWithAttachedContainer)
- .create())
- } catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to create executor pod ${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- None
+ }
+ }
+
+ /**
+ * Parallel pod creation: builds pod specs serially (they depend on shared mutable state like
+ * EXECUTOR_ID_COUNTER and reusablePVCs), then creates them via K8s API in parallel using a
+ * thread pool sized by `spark.kubernetes.allocation.batch.concurrency`.
+ */
+ private def requestNewExecutorsParallel(
+ numExecutorsToAllocate: Int,
+ applicationId: String,
+ resourceProfileId: Int,
+ pvcsInUse: Seq[String]): Unit = {
+ val batchStartTime = clock.getTimeMillis()
+
+ // Check reusable PVCs for this executor allocation batch
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+
+ // Phase 1: Build all pod specs serially
+ val podSpecs = mutable.ArrayBuffer.empty[(Long, Pod, Seq[HasMetadata])]
+ var i = 0
+ while (i < numExecutorsToAllocate) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <= PVC_COUNTER.get()) {
+ logInfo(
+ log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT, PVC_COUNTER.get())} PVCs.")
+ // Stop building more specs; proceed to create whatever we've built so far
+ i = numExecutorsToAllocate
+ } else {
+ podSpecs += buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ i += 1
}
- optCreatedExecutorPod.foreach { createdExecutorPod =>
+ }
+ if (podSpecs.isEmpty) return
+
+ val podBuildTotalMs = clock.getTimeMillis() - batchStartTime
+
+ // Phase 2: Create pods via K8s API in parallel
+ val k8sApiStartTime = clock.getTimeMillis()
+
+ implicit val ec: ExecutionContext = podCreationExecutionContext
+
+ val futures = podSpecs.map { case (newExecutorId, podWithAttachedContainer, resources) =>
+ Future {
try {
- addOwnerReference(createdExecutorPod, resources)
- resources
- .filter(_.getKind == "PersistentVolumeClaim")
- .foreach { resource =>
- if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
- addOwnerReference(driverPod.get, Seq(resource))
- }
- val pvc = resource.asInstanceOf[PersistentVolumeClaim]
- logInfo(log"Trying to create PersistentVolumeClaim " +
- log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)} with " +
- log"StorageClass ${MDC(LogKeys.CLASS_NAME, pvc.getSpec.getStorageClassName)}")
- kubernetesClient
- .persistentVolumeClaims()
- .inNamespace(namespace)
- .resource(pvc)
- .create()
- PVC_COUNTER.incrementAndGet()
- }
- newlyCreatedExecutors(newExecutorId) = (resourceProfileId, clock.getTimeMillis())
- logDebug(s"Requested executor with id $newExecutorId from Kubernetes.")
+ val result = createExecutorPodAndPVCs(newExecutorId, podWithAttachedContainer, resources)
+ result.map(execId => (execId, resourceProfileId, clock.getTimeMillis()))
} catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to add owner reference or create PVC for executor pod " +
- log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- kubernetesClient.pods()
- .inNamespace(namespace)
- .resource(createdExecutorPod)
- .delete()
- throw e
+ // createExecutorPodAndPVCs already logs the error and registers the failure
+ // before rethrowing. We catch here only to prevent a failed Future from
+ // short-circuiting Future.sequence, so that other concurrent pods can proceed.
+ case NonFatal(_) => None
}
}
}
+
+ val results = try {
+ val allResults = ThreadUtils.awaitResult(
+ Future.sequence(futures), Duration(podCreationTimeout, TimeUnit.MILLISECONDS))
+ allResults.flatten
+ } catch {
+ case NonFatal(e) =>
+ logError(log"Timed out waiting for parallel pod creation to complete " +
+ log"after ${MDC(LogKeys.TIMEOUT, podCreationTimeout)} ms", e)
+ Seq.empty
Review Comment:
> If one creation action times out, none of the Pods in this batch will be considered successfully created. this should **not** be batch-atomic, right?
Yes! This should not be batch-atomic. I've updated the code to await each future individually instead of using Future.sequence.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55888][K8S] Parallel pod creation for Kubernetes executor allocation [spark]
Posted by "yikf (via GitHub)" <gi...@apache.org>.
yikf commented on code in PR #54688:
URL: https://github.com/apache/spark/pull/54688#discussion_r2924382798
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -574,6 +711,11 @@ class ExecutorPodsAllocator(
}
override def stop(applicationId: String): Unit = {
+ if (podAllocationConcurrency > 1) {
+ Utils.tryLogNonFatalError {
+ podCreationThreadPool.shutdownNow()
Review Comment:
Even if `snapshotsStore.stop()` is invoked first, it does not guarantee thread `subscribersExecutor` stops first.
A thread pool's shutdown does not terminate threads that do not respond to interrupts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org