peter-toth commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3824198853
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,232 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors and can hold them,
decommission support,
+ * and shuffle data kept outside the executors: either an external shuffle
service or a
+ * `ShuffleDataIO` with reliable storage.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ (schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend => cg.supportsExecutorHold
+ case _ => false
+ }) &&
+ (conf.get(SHUFFLE_SERVICE_ENABLED) ||
shuffleDriverComponents.supportsReliableStorage()) &&
+ conf.get(DECOMMISSION_ENABLED)
+ }
+
+ /** Whether the executors are currently held by `holdExecutors()`. */
+ private[spark] def executorsHeld: Boolean = _executorsHeld
+
+ /**
+ * :: DeveloperApi ::
+ * Hold the whole application by declining to allocate new executors and
gracefully
+ * decommissioning all existing ones. Each executor finishes its running
tasks and then
+ * exits -- unless `spark.executor.decommission.forceKillTimeout` is set, in
which case an
+ * executor still running tasks is killed after that timeout. The shuffle
data already
+ * written remains available outside the executors, so the application can
later pick up
+ * where it left off via `resumeExecutors()`. Cached blocks are not
preserved and are
+ * recomputed after resuming. While held the application has no executors,
so with
+ * `spark.default.parallelism` unset the default parallelism falls back to
2, and an RDD
+ * created during the hold keeps that partition count after resuming.
+ *
+ * This requires decommission support (`spark.decommission.enabled`),
shuffle data kept
+ * outside the executors -- either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage -- and a
+ * scheduler backend that can hold executors: Standalone, YARN, and
Kubernetes with the
+ * `direct` pods allocator. Fallback storage
+ * (`spark.storage.decommission.fallbackStorage.path`) deliberately does not
qualify:
+ * shuffle blocks not yet migrated when an executor exits are dropped.
+ *
+ * Executor requirements requested while held, through `requestExecutors` or
+ * `requestTotalExecutors`, are recorded but nothing is allocated until
`resumeExecutors()`
+ * restores them.
+ *
+ * Pipelined-shuffle jobs are outside the hold's scope. A hold is rejected,
on a
+ * best-effort check, while a pipelined job is running: its transient
shuffle data lives
+ * only on the executors and would not survive the drain, and a group that
slips past the
+ * check is aborted rather than drained, since a pipelined task set
tolerates no task
+ * failure. A pipelined job submitted while held fails its gang admission
immediately
Review Comment:
**Finding 38.** With
`spark.scheduler.pipelinedGroup.slotCheck.enabled=false`,
`rejectUnadmittablePipelinedGroup` returns `false` before it ever compares
demand against free slots
(`core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:1872-1883`).
So in that deployment a pipelined job submitted while held *is* admitted: its
stages get created and its task sets sit in the scheduler with no executors
until the resume. It waits — which is what this commit removed for the default
config. "Resubmit it after the resume" then costs a needless resubmit of a job
that was going to run.
The config is `internal()`, so this is small. One clause covers it:
```scala
* failure. A pipelined job submitted while held fails its gang admission
immediately
* instead of waiting; resubmit it after the resume (with
* `spark.scheduler.pipelinedGroup.slotCheck.enabled=false` there is no
admission check,
* so it waits for the resume instead). Workloads with long-running tasks
(a
```
Same sentence in `docs/web-ui.md:82-83` and in the PR description.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
