dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3815694834
##########
docs/web-ui.md:
##########
@@ -70,6 +70,16 @@ The information displayed at the top of the page includes:
The current user, application start time, and total uptime are shown in the
footer at the
bottom of every page.
+When the application can be held, the summary shows an **Application** line
with a **(hold)**
+link; clicking it stops requesting new executors and gracefully decommissions
the running ones,
+so each finishes its tasks and then exits. The line then reads `Held` with a
**(resume)** link
+that restores the executor requirement. Shuffle output written before the hold
stays available,
+but cached blocks are recomputed after resuming. A hold requested while a
pipelined job is
Review Comment:
Documented in a927b1e, in the `holdExecutors()` scaladoc and next to the
cached-blocks sentence in `web-ui.md`: 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.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,213 @@ 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,
+ * while 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.
+ *
+ * This requires decommission support (`spark.decommission.enabled`) and
shuffle data kept
+ * outside the executors: either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage.
+ *
+ * Executor requirements requested while held, through `requestExecutors` or
+ * `requestTotalExecutors`, are recorded but nothing is allocated until
`resumeExecutors()`
+ * restores them.
+ *
+ * The 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.
+ *
+ * @return whether the lowered executor requirement was acknowledged by the
cluster manager.
+ * With dynamic allocation a rejected request is retried in the
background; the
+ * executors are drained in either case.
+ */
+ @DeveloperApi
+ def holdExecutors(): Boolean = {
+ schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend if cg.supportsExecutorHold =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
+ s"${SHUFFLE_SERVICE_ENABLED.key} or a ShuffleDataIO with reliable
storage")
+ val pipelinedRunning = taskScheduler match {
+ case ts: TaskSchedulerImpl => ts.hasPipelinedTaskSets
+ case _ => false
+ }
+ if (pipelinedRunning) {
+ // A pipelined group reads and writes transient shuffle data that
lives only on
+ // its executors: a partially launched group would deadlock the
drain, and a
+ // force-killed member aborts the whole group.
+ logWarning(log"Cannot hold the executors while a pipelined job is
running.")
+ false
+ } else synchronized {
+ if (_executorsHeld) {
+ // A repeated hold re-asserts the zero requirement: the earlier
publish may not
+ // have been acknowledged, and with dynamic allocation off nothing
retries it.
+ if (executorAllocationManager.isDefined) true else
zeroExecutorRequirementAndDrain(cg)
+ } else {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume when none was explicitly
requested
+ // (explicitly requested totals, made before or during the hold,
are
+ // republished from the backend directly). Only killExecutors'
bookkeeping
+ // zero is kill-seeded (read atomically against a concurrent
reset): restore
+ // the count of executors not already being removed, so that
resume neither
+ // parks the application at zero nor undoes the downscale.
Otherwise
+ // Standalone has no explicit requirement by default (and ignores
+ // spark.executor.instances, even a leftover value), so restore
an unbounded
+ // one; elsewhere follow the conf, or fall back to the cluster
manager's
+ // default when no executor has registered yet.
+ heldNumExecutors = if (cg.hasKillSeededTotalsOnly) {
+ cg.activeExecutorCount
+ } else {
+ schedulerBackend match {
+ case _: StandaloneSchedulerBackend => Int.MaxValue
Review Comment:
Fixed in a927b1e with the simpler variant: `updateExecRequestTimes` skips
the unbounded sentinel, so it never reaches the request-time queue. I stayed
away from the tidier clear-and-publish route: after a hold the Master-side
executor limit is 0, and publishing an empty map does not restore it on
standalone (`ApplicationInfo` treats an empty map as a no-op), so "forget the
requirement" would leave the application parked at zero -- the sentinel through
`requestTotalExecutors` is what actually lifts the limit.
--
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]
