You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/03/17 23:53:44 UTC
(spark) branch master updated: [SPARK-55963][K8S] Optimize snapshot traversal in ExecutorPodsAllocator
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 523921abc17c [SPARK-55963][K8S] Optimize snapshot traversal in ExecutorPodsAllocator
523921abc17c is described below
commit 523921abc17c071bfc8ae2bcf76d85de722a7a14
Author: DenineLu <de...@163.com>
AuthorDate: Tue Mar 17 16:53:35 2026 -0700
[SPARK-55963][K8S] Optimize snapshot traversal in ExecutorPodsAllocator
### What changes were proposed in this pull request?
Optimize the snapshot preprocessing in `ExecutorPodsAllocator.onNewSnapshots()`.
Currently, to collect known executor IDs and PVC names, the code calls `flatMap` + `distinct` across all snapshots. Since each snapshot is built incrementally via `withUpdate()`, they overlap heavily — when requesting 1000 pods, a single batch may contain hundreds of snapshots, each carrying a near-complete copy of the pod map. This leads to a lot of redundant work.
This PR merges all snapshots into a single aggregated map up front, which deduplicates by executor ID naturally. PVC names are then extracted from this deduplicated map. The merge is still correct in the presence of `replaceSnapshot()`, which can drop pods from earlier snapshots. Also changed `pvcsInUse` from `Seq` to `Set` for O(1) `contains()`.
**Note:** This PR is part of a broader effort to improve K8s executor allocation performance. It targets the **snapshot preprocessing** phase of `onNewSnapshots()`. A companion PR #54688 addresses the other major bottleneck — **serial pod creation** — by parallelizing K8s API calls. The two optimizations are independent and complementary.
### Why are the changes needed?
When requesting 1000 executor pods, the average preprocessing time (`totalPreprocessMs`) dropped from **17.6 ms to 1.5 ms** (~11.7x speedup). This overhead is incurred on every snapshot batch and can delay executor allocation at scale.
<img width="1430" height="440" alt="image" src="https://github.com/user-attachments/assets/c095d460-4748-4ac3-b94f-d301a8c44cfc" />
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Updated existing test
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #54759 from DenineLu/optimize-pod-allocator-checker.
Authored-by: DenineLu <de...@163.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../cluster/k8s/ExecutorPodsAllocator.scala | 47 ++++++++++++++++++----
.../cluster/k8s/ExecutorPodsAllocatorSuite.scala | 5 ++-
2 files changed, 43 insertions(+), 9 deletions(-)
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
index 9322970e6b98..c20346edc840 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
@@ -165,18 +165,51 @@ class ExecutorPodsAllocator(
applicationId: String,
schedulerBackend: KubernetesClusterSchedulerBackend,
snapshots: Seq[ExecutorPodsSnapshot]): Unit = {
+ val snapshotProcessStartTime = clock.getTimeMillis()
logDebug(s"Received ${snapshots.size} snapshots")
- val k8sKnownExecIds = snapshots.flatMap(_.executorPods.keys).distinct
+
+ // Optimization: Since each snapshot is built incrementally via withUpdate() on top of the
+ // previous one, the last snapshot is always a superset of all prior snapshots in most cases.
+ // However, replaceSnapshot() can cause pods that appeared in earlier snapshots to be absent
+ // in the last one. We merge all snapshots to capture every executor ID that appeared during
+ // this batch, which is the conservative and correct approach.
+ val aggregatedPods: Map[Long, ExecutorPodState] = if (snapshots.size <= 1) {
+ snapshots.headOption.map(_.executorPods).getOrElse(Map.empty)
+ } else {
+ val merged = mutable.HashMap.empty[Long, ExecutorPodState]
+ snapshots.foreach { s => merged ++= s.executorPods }
+ merged.toMap
+ }
+
+ val k8sKnownExecIds = aggregatedPods.keySet
newlyCreatedExecutors --= k8sKnownExecIds
schedulerKnownNewlyCreatedExecs --= k8sKnownExecIds
// Although we are going to delete some executors due to timeout in this function,
// it takes undefined time before the actual deletion. Hence, we should collect all PVCs
// in use at the beginning. False positive is okay in this context in order to be safe.
- val k8sKnownPVCNames = snapshots.flatMap(_.executorPods.values.map(_.pod)).flatMap { pod =>
- pod.getSpec.getVolumes.asScala
- .flatMap { v => Option(v.getPersistentVolumeClaim).map(_.getClaimName) }
- }.distinct
+ // Optimization: iterate over the aggregated pod set (already deduplicated by exec ID)
+ // instead of iterating all snapshots which contain massive overlap.
+ val k8sKnownPVCNames: Set[String] = {
+ val pvcNameSet = mutable.HashSet.empty[String]
+ aggregatedPods.valuesIterator.foreach { state =>
+ state.pod.getSpec.getVolumes.asScala.foreach { v =>
+ val pvc = v.getPersistentVolumeClaim
+ if (pvc != null) {
+ pvcNameSet.add(pvc.getClaimName)
+ }
+ }
+ }
+ pvcNameSet.toSet
+ }
+ val snapshotPreprocessMs = clock.getTimeMillis() - snapshotProcessStartTime
+ if (snapshots.size > 1) {
+ logDebug(s"Snapshot preprocessing: " +
+ s"snapshotCount=${snapshots.size}, " +
+ s"aggregatedPodCount=${aggregatedPods.size}, " +
+ s"pvcCount=${k8sKnownPVCNames.size}, " +
+ s"totalPreprocessMs=${snapshotPreprocessMs}")
+ }
// transfer the scheduler backend known executor requests from the newlyCreatedExecutors
// to the schedulerKnownNewlyCreatedExecs
@@ -399,7 +432,7 @@ class ExecutorPodsAllocator(
numOutstandingPods.set(totalPendingCount + newlyCreatedExecutors.size)
}
- protected def getReusablePVCs(applicationId: String, pvcsInUse: Seq[String]) = {
+ protected def getReusablePVCs(applicationId: String, pvcsInUse: Set[String]) = {
if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && conf.get(KUBERNETES_DRIVER_REUSE_PVC) &&
driverPod.nonEmpty) {
try {
@@ -437,7 +470,7 @@ class ExecutorPodsAllocator(
numExecutorsToAllocate: Int,
applicationId: String,
resourceProfileId: Int,
- pvcsInUse: Seq[String]): Unit = {
+ pvcsInUse: Set[String]): Unit = {
// Check reusable PVCs for this executor allocation batch
val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
for ( _ <- 0 until numExecutorsToAllocate) {
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
index 3f2e344b1c23..843e704e5c01 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
@@ -899,7 +899,7 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
val getReusablePVCs =
PrivateMethod[mutable.Buffer[PersistentVolumeClaim]](Symbol("getReusablePVCs"))
when(persistentVolumeClaimList.getItems).thenThrow(new KubernetesClientException("Error"))
- podsAllocatorUnderTest invokePrivate getReusablePVCs("appId", Seq.empty[String])
+ podsAllocatorUnderTest invokePrivate getReusablePVCs("appId", Set.empty[String])
}
test("SPARK-41388: getReusablePVCs should ignore recently created PVCs in the previous batch") {
@@ -914,7 +914,8 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
pvc2.getMetadata.setCreationTimestamp(now.toString)
when(persistentVolumeClaimList.getItems).thenReturn(Seq(pvc1, pvc2).asJava)
- val reusablePVCs = podsAllocatorUnderTest invokePrivate getReusablePVCs("appId", Seq.empty)
+ val reusablePVCs =
+ podsAllocatorUnderTest invokePrivate getReusablePVCs("appId", Set.empty[String])
assert(reusablePVCs.size == 1)
assert(reusablePVCs.head.getMetadata.getName == "pvc-1")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org