[SPARK-56693][K8S] Support built-in K8s ExecutorPVCResizePlugin

dongjoon-hyun · dongjoon-hyun · commit 8e3782440253 · 2026-05-01T00:34:12.000-07:00
### What changes were proposed in this pull request? This PR aims to support a new `ExecutorPVCResizePlugin` that monitors executor PVC disk usage and grows each PVC's `spec.resources.requests.storage` when usage exceeds a threshold. The executor side reports the max filesystem usage ratio across `DiskBlockManager.localDirs`. The driver side patches the executor pod's PVCs to `currentSize * (1 + factor)` when the reported ratio exceeds the threshold. New configurations: | Key | Default | Meaning | |---|---|---| | `spark.kubernetes.executor.pvc.resizeInterval` | `0min` | Resize check interval. `0` disables. | | `spark.kubernetes.executor.pvc.resizeThreshold` | `0.5` | Usage ratio above which a resize is triggered. | | `spark.kubernetes.executor.pvc.resizeFactor` | `1.0` | Growth factor. | Note that the public cloud PVC has clear limitations. For example, you can increase only and **once per every six hour**. So, for a short Spark job (whose lifetime is less than 6 hours), `ExecutorPVCResizePlugin` is able to increase only one time. For a recently increased PVCs, K8s will show a warning something like the following. ``` Warning VolumeResizeFailed 6s (x8 over 35s) external-resizer ebs.csi.aws.com resize volume "pvc-baccb8c8-16e7-4b31-b254-dd223a3e0be3" by resizer "ebs.csi.aws.com" failed: rpc error: code = Internal desc = Could not resize volume "vol-0cb3afad24b360ebd": rpc error: code = Internal desc = Could not modify vol ume "vol-0cb3afad24b360ebd": volume "vol-0cb3afad24b360ebd" in OPTIMIZING state, cannot currently modify ``` ### Why are the changes needed? PVC-backed `SPARK_LOCAL_DIRS` must be sized conservatively up front to avoid mid-job disk-full failures, which wastes storage cost. `ExecutorResizePlugin` already established the observe-and-patch pattern for memory; this extends it to PVC storage. ### Does this PR introduce _any_ user-facing change? No. The user needs to set this to `spark.plugins` explicitly. **SUBMIT** ``` bin/spark-submit \ --master k8s://$K8S_MASTER \ --deploy-mode cluster \ -c spark.executor.cores=4 \ -c spark.executor.memory=4g \ -c spark.kubernetes.container.image=docker.apple.com/d_hyun/spark:20260430 \ -c spark.kubernetes.authenticate.driver.serviceAccountName=spark \ -c spark.kubernetes.driver.pod.name=pi \ -c spark.kubernetes.executor.podNamePrefix=pi \ -c spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.mount.path=/data \ -c spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.mount.readOnly=false \ -c spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.claimName=OnDemand \ -c spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.sizeLimit=50Gi \ -c spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.storageClass=gp3 \ -c spark.kubernetes.driver.podTemplateFile=eks-root-pod.yml \ -c spark.kubernetes.executor.podTemplateFile=eks-root-pod.yml \ -c spark.plugins=org.apache.spark.scheduler.cluster.k8s.ExecutorPVCResizePlugin \ -c spark.kubernetes.executor.pvc.resizeInterval=1m \ --class org.apache.spark.examples.SparkPi \ local:///opt/spark/examples/jars/spark-examples.jar 400000 ``` **DRIVER LOGS** ``` $ kubectl logs -f pi | grep Plugin 26/05/01 01:39:38 INFO ExecutorPVCResizeDriverPlugin: ExecutorPVCResizeDriverPlugin is scheduled 26/05/01 01:39:38 INFO DriverPluginContainer: Initialized driver component for plugin org.apache.spark.scheduler.cluster.k8s.ExecutorPVCResizePlugin. 26/05/01 01:40:38 INFO ExecutorPVCResizeDriverPlugin: Latest PVC usage reports: {} 26/05/01 01:41:38 INFO ExecutorPVCResizeDriverPlugin: Latest PVC usage reports: {1=PVCDiskUsageReport(1,0.6136656796630462), 2=PVCDiskUsageReport(2,3.507855787665699E-4)} 26/05/01 01:41:38 INFO ExecutorPVCResizeDriverPlugin: Try to resize executor 1 PVC pi-exec-1-pvc-0 with ratio 0.6136656796630462 (threshold 0.5). 26/05/01 01:41:38 INFO ExecutorPVCResizeDriverPlugin: Increase PVC pi-exec-1-pvc-0 storage from 53687091200 to 107374182400 as usage ratio exceeded threshold. 26/05/01 01:41:38 INFO ExecutorPVCResizeDriverPlugin: Try to resize executor 2 PVC pi-exec-2-pvc-0 with ratio 3.507855787665699E-4 (threshold 0.5). ``` **EXECUTOR SIZE REPORTING (60% -> 30%)** ``` $ kubectl logs -f pi-exec-1 | grep Plugin 26/05/01 01:22:54 INFO ExecutorPVCResizeExecutorPlugin: Reporting max PVC disk usage ratio for executor 1: 0.6136656796630462 26/05/01 01:23:54 INFO ExecutorPVCResizeExecutorPlugin: Reporting max PVC disk usage ratio for executor 1: 0.30591566408202353 ``` **RESIZED PVC** ``` $ kubectl get pvc NAME STATUS VOLUME CAPACITY ACCESS MODES STORAGECLASS VOLUMEATTRIBUTESCLASS AGE pi-exec-1-pvc-0 Bound pvc-d279a3da-ddfb-41c2-a32b-0f2bd83941c4 107374182400 RWOP gp3 <unset> 2m28s pi-exec-2-pvc-0 Bound pvc-79f092d3-4a8d-4981-946d-d745d4038fd6 50Gi RWOP gp3 <unset> 2m28s ``` ### How was this patch tested? Pass the CIs with a new `ExecutorPVCResizePluginSuite`. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 4.7 (1M context) Closes #55642 from dongjoon-hyun/SPARK-56693. Authored-by: Dongjoon Hyun <dongjoon@apache.org> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
@@ -1816,6 +1816,36 @@ See the [configuration page](configuration.html) for information on Spark config
   </td>
   <td>4.2.0</td>
 </tr>
+<tr>
+  <td><code>spark.kubernetes.executor.pvc.resizeInterval</code></td>
+  <td><code>0min</code></td>
+  <td>
+    Interval between executor PVC resize operations. To disable, set 0 (default).
+    Takes effect only when <code>org.apache.spark.scheduler.cluster.k8s.ExecutorPVCResizePlugin</code>
+    is registered via <code>spark.plugins</code>.
+  </td>
+  <td>4.2.0</td>
+</tr>
+<tr>
+  <td><code>spark.kubernetes.executor.pvc.resizeThreshold</code></td>
+  <td><code>0.5</code></td>
+  <td>
+    The PVC usage ratio (used / capacity) above which the driver triggers a resize.
+    Takes effect only when <code>org.apache.spark.scheduler.cluster.k8s.ExecutorPVCResizePlugin</code>
+    is registered via <code>spark.plugins</code>.
+  </td>
+  <td>4.2.0</td>
+</tr>
+<tr>
+  <td><code>spark.kubernetes.executor.pvc.resizeFactor</code></td>
+  <td><code>1.0</code></td>
+  <td>
+    The factor to grow PVC storage by, relative to the current request.
+    Takes effect only when <code>org.apache.spark.scheduler.cluster.k8s.ExecutorPVCResizePlugin</code>
+    is registered via <code>spark.plugins</code>.
+  </td>
+  <td>4.2.0</td>
+</tr>
 </table>
 
 #### Pod template properties
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
@@ -262,6 +262,30 @@ private[spark] object Config extends Logging {
       .checkValue(v => 0 < v && v <= 1, "The factor should be in (0, 1]")
       .createWithDefault(0.1)
 
+  val PVC_RESIZE_INTERVAL =
+    ConfigBuilder("spark.kubernetes.executor.pvc.resizeInterval")
+      .doc("Interval between executor PVC resize operations. To disable, set 0 (default)")
+      .version("4.2.0")
+      .timeConf(TimeUnit.MINUTES)
+      .checkValue(_ >= 0, "Interval should be non-negative")
+      .createWithDefault(0)
+
+  val PVC_RESIZE_THRESHOLD =
+    ConfigBuilder("spark.kubernetes.executor.pvc.resizeThreshold")
+      .doc("The PVC usage ratio (used / capacity) above which the driver triggers a resize.")
+      .version("4.2.0")
+      .doubleConf
+      .checkValue(v => 0 < v && v < 1, "The threshold should be in (0, 1)")
+      .createWithDefault(0.5)
+
+  val PVC_RESIZE_FACTOR =
+    ConfigBuilder("spark.kubernetes.executor.pvc.resizeFactor")
+      .doc("The factor to grow PVC storage by, relative to the current request.")
+      .version("4.2.0")
+      .doubleConf
+      .checkValue(v => 0 < v && v <= 1, "The factor should be in (0, 1]")
+      .createWithDefault(1.0)
+
   val KUBERNETES_AUTH_DRIVER_CONF_PREFIX = "spark.kubernetes.authenticate.driver"
   val KUBERNETES_AUTH_EXECUTOR_CONF_PREFIX = "spark.kubernetes.authenticate.executor"
   val KUBERNETES_AUTH_DRIVER_MOUNTED_CONF_PREFIX = "spark.kubernetes.authenticate.driver.mounted"
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePlugin.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePlugin.scala
@@ -0,0 +1,255 @@
+/*
+ * 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.util.{Map => JMap}
+import java.util.concurrent.{ConcurrentHashMap, ScheduledExecutorService, TimeUnit}
+
+import scala.jdk.CollectionConverters._
+
+import io.fabric8.kubernetes.api.model.{PersistentVolumeClaimBuilder, Pod, Quantity}
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.client.dsl.base.PatchContext
+import io.fabric8.kubernetes.client.dsl.base.PatchType
+
+import org.apache.spark.{SparkContext, SparkEnv}
+import org.apache.spark.api.plugin.{DriverPlugin, ExecutorPlugin, PluginContext, SparkPlugin}
+import org.apache.spark.deploy.k8s.Config._
+import org.apache.spark.deploy.k8s.Constants._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.{CURRENT_DISK_SIZE, ORIGINAL_DISK_SIZE, PVC_METADATA_NAME}
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Spark plugin to monitor executor PVC disk usage and grow the PVC storage request
+ * when the usage exceeds a configurable threshold.
+ *
+ * Executors measure their own local-directory usage (via DiskBlockManager) and report
+ * the maximum filesystem usage ratio to the driver through the plugin RPC channel.
+ * When the ratio exceeds the threshold, the driver patches every `spark-local-dir-*`
+ * PVC mounted by the reporting executor's pod to grow its
+ * `spec.resources.requests.storage`. The underlying StorageClass must have
+ * `allowVolumeExpansion: true`.
+ */
+class ExecutorPVCResizePlugin extends SparkPlugin {
+  override def driverPlugin(): DriverPlugin = new ExecutorPVCResizeDriverPlugin()
+
+  override def executorPlugin(): ExecutorPlugin = new ExecutorPVCResizeExecutorPlugin()
+}
+
+/**
+ * Message sent from each executor to the driver with the maximum filesystem usage
+ * ratio (used / total) across the executor's SPARK_LOCAL_DIRS. The driver applies
+ * this ratio to every PVC mounted by the reporting executor's pod.
+ */
+private[k8s] case class PVCDiskUsageReport(
+    executorId: String,
+    ratio: Double)
+
+class ExecutorPVCResizeDriverPlugin extends DriverPlugin with Logging {
+  private var sparkContext: SparkContext = _
+  private var namespace: String = _
+  private var threshold: Double = _
+  private var factor: Double = _
+
+  private val latestReports = new ConcurrentHashMap[String, PVCDiskUsageReport]()
+  private val failedPvcs = ConcurrentHashMap.newKeySet[String]()
+  private val requestedSizes = new ConcurrentHashMap[String, Long]()
+
+  private val periodicService: ScheduledExecutorService =
+    ThreadUtils.newDaemonSingleThreadScheduledExecutor("pvc-resize-plugin")
+
+  override def init(sc: SparkContext, ctx: PluginContext): JMap[String, String] = {
+    val interval = sc.conf.get(PVC_RESIZE_INTERVAL)
+    if (interval <= 0) {
+      logInfo("PVCResizePlugin disabled (interval <= 0).")
+      return Map.empty[String, String].asJava
+    }
+    threshold = sc.conf.get(PVC_RESIZE_THRESHOLD)
+    factor = sc.conf.get(PVC_RESIZE_FACTOR)
+    namespace = sc.conf.get(KUBERNETES_NAMESPACE)
+    sparkContext = sc
+
+    periodicService.scheduleAtFixedRate(() => if (!sparkContext.isStopped) {
+      try {
+        checkAndResizePVCs()
+      } catch {
+        case e: Throwable => logError("Error in PVC resize thread", e)
+      }
+    }, interval, interval, TimeUnit.MINUTES)
+    logInfo("ExecutorPVCResizeDriverPlugin is scheduled")
+
+    // Propagate the interval to executors so they report at the same cadence.
+    Map(PVC_RESIZE_INTERVAL.key -> interval.toString).asJava
+  }
+
+  override def receive(message: Any): AnyRef = message match {
+    case r: PVCDiskUsageReport =>
+      latestReports.put(r.executorId, r)
+      null
+    case _ =>
+      null
+  }
+
+  override def shutdown(): Unit = {
+    periodicService.shutdown()
+  }
+
+  private[k8s] def checkAndResizePVCs(): Unit = {
+    logInfo(s"Latest PVC usage reports: $latestReports")
+    val appId = sparkContext.applicationId
+
+    sparkContext.schedulerBackend match {
+      case b: KubernetesClusterSchedulerBackend =>
+        val client = b.kubernetesClient
+        val pods = client.pods()
+          .inNamespace(namespace)
+          .withLabel(SPARK_APP_ID_LABEL, appId)
+          .withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
+          .list()
+          .getItems.asScala
+
+        val podByExecId = pods.flatMap { p =>
+          Option(p.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL)).map(_ -> p)
+        }.toMap
+
+        latestReports.values().asScala.foreach { report =>
+          podByExecId.get(report.executorId).foreach { pod =>
+            pvcsOf(pod).foreach { pvcName =>
+              if (!failedPvcs.contains(pvcName)) {
+                tryResize(client, pvcName, report.ratio, report.executorId)
+              }
+            }
+          }
+        }
+      case _ =>
+        logWarning("Skipping PVC resize: schedulerBackend is not " +
+          "KubernetesClusterSchedulerBackend.")
+    }
+  }
+
+  private[k8s] def pvcsOf(pod: Pod): Set[String] = {
+    val volNameToPvc = pod.getSpec.getVolumes.asScala
+      .filter(_.getPersistentVolumeClaim != null)
+      .filter(_.getName.startsWith("spark-local-dir-"))
+      .map(v => v.getName -> v.getPersistentVolumeClaim.getClaimName)
+      .toMap
+    pod.getSpec.getContainers.asScala
+      .find(_.getName == DEFAULT_EXECUTOR_CONTAINER_NAME)
+      .orElse(pod.getSpec.getContainers.asScala.headOption)
+      .toSeq
+      .flatMap(_.getVolumeMounts.asScala)
+      .flatMap(m => volNameToPvc.get(m.getName))
+      .toSet
+  }
+
+  private def tryResize(
+      client: KubernetesClient, pvcName: String, ratio: Double, execId: String): Unit = {
+    logInfo(s"Try to resize executor $execId PVC $pvcName with ratio $ratio " +
+      s"(threshold $threshold).")
+    if (ratio <= threshold) return
+    try {
+      val pvc = client.persistentVolumeClaims()
+        .inNamespace(namespace)
+        .withName(pvcName)
+        .get()
+      if (pvc == null) return
+      val current = Quantity.getAmountInBytes(
+        pvc.getSpec.getResources.getRequests.get("storage")).longValue()
+      val capacity = Option(pvc.getStatus)
+        .flatMap(s => Option(s.getCapacity))
+        .flatMap(c => Option(c.get("storage")))
+        .map(q => Quantity.getAmountInBytes(q).longValue())
+        .getOrElse(current)
+      if (current > capacity) {
+        logInfo(s"PVC $pvcName resize is in progress or failed " +
+          s"(spec=$current, status=$capacity); skip.")
+        return
+      }
+      val newSize = (current * (1.0 + factor)).toLong
+      if (requestedSizes.get(pvcName) == newSize) return
+      logInfo(log"Increase PVC ${MDC(PVC_METADATA_NAME, pvcName)} storage " +
+        log"from ${MDC(ORIGINAL_DISK_SIZE, current)} to " +
+        log"${MDC(CURRENT_DISK_SIZE, newSize)} as usage ratio exceeded threshold.")
+      val patch = new PersistentVolumeClaimBuilder()
+        .withNewSpec()
+          .withNewResources()
+            .addToRequests("storage", new Quantity(newSize.toString))
+          .endResources()
+        .endSpec()
+        .build()
+      client.persistentVolumeClaims()
+        .inNamespace(namespace)
+        .withName(pvcName)
+        .patch(PatchContext.of(PatchType.STRATEGIC_MERGE), patch)
+      requestedSizes.put(pvcName, newSize)
+    } catch {
+      case e: Throwable =>
+        failedPvcs.add(pvcName)
+        logInfo(log"Failed to expand PVC ${MDC(PVC_METADATA_NAME, pvcName)}; " +
+          log"will skip subsequent attempts.", e)
+    }
+  }
+}
+
+class ExecutorPVCResizeExecutorPlugin extends ExecutorPlugin with Logging {
+  private var pluginContext: PluginContext = _
+  private var periodicService: ScheduledExecutorService = _
+
+  override def init(ctx: PluginContext, extraConf: JMap[String, String]): Unit = {
+    val intervalStr = extraConf.get(PVC_RESIZE_INTERVAL.key)
+    if (intervalStr == null) {
+      // Driver disabled the plugin; do nothing.
+      return
+    }
+    val interval = intervalStr.toLong
+    if (interval <= 0) return
+
+    pluginContext = ctx
+    periodicService =
+      ThreadUtils.newDaemonSingleThreadScheduledExecutor("pvc-resize-reporter")
+    periodicService.scheduleAtFixedRate(() => {
+      try {
+        report()
+      } catch {
+        case e: Throwable => logDebug("Failed to report PVC usage", e)
+      }
+    }, interval, interval, TimeUnit.MINUTES)
+  }
+
+  override def shutdown(): Unit = {
+    if (periodicService != null) periodicService.shutdown()
+  }
+
+  private def report(): Unit = {
+    val env = SparkEnv.get
+    if (env == null) return
+    val dirs = env.blockManager.diskBlockManager.localDirs
+    if (dirs == null || dirs.isEmpty) return
+    val maxRatio = dirs.iterator.flatMap { d =>
+      try {
+        // Skip if total is 0 (e.g. dir unmounted, statvfs failed) to avoid divide-by-zero.
+        val total = d.getTotalSpace
+        if (total > 0) Some((total - d.getUsableSpace).toDouble / total) else None
+      } catch { case _: Throwable => None }
+    }.maxOption
+    maxRatio.foreach { ratio =>
+      logInfo(s"Reporting max PVC disk usage ratio for executor ${env.executorId}: $ratio")
+      pluginContext.send(PVCDiskUsageReport(env.executorId, ratio))
+    }
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePluginSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePluginSuite.scala
