dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3805778839
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && 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 through the
external shuffle
+ * service, so the application can later pick up where it left off via
`resumeExecutors()`.
+ *
+ * This requires both an external shuffle service
(`spark.shuffle.service.enabled`) and
+ * decommission support (`spark.decommission.enabled`).
+ *
+ * @return whether the request is received by the cluster manager.
+ */
+ @DeveloperApi
+ def holdExecutors(): Boolean = {
+ schedulerBackend match {
+ case b: ExecutorAllocationClient =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires both ${SHUFFLE_SERVICE_ENABLED.key} and "
+
+ s"${DECOMMISSION_ENABLED.key}")
+ synchronized {
+ if (!_executorsHeld) {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume. Standalone ignores
+ // spark.executor.instances and has no explicit executor
requirement by default,
+ // so restore an unbounded one, even when the conf carries a
leftover value.
+ // Elsewhere follow the conf, or fall back to the cluster
manager's default when
+ // no executor has registered yet.
+ heldNumExecutors = schedulerBackend match {
+ case _: StandaloneSchedulerBackend => Int.MaxValue
+ case _ =>
+
conf.get(EXECUTOR_INSTANCES).getOrElse(math.max(b.getExecutorIds().size,
+ SchedulerBackendUtils.DEFAULT_NUMBER_EXECUTORS))
+ }
+ }
+ // Mark the hold before talking to the cluster manager, so that a
partial failure
+ // below leaves the executors held, and thus resumable, instead of
half-held.
+ _executorsHeld = true
+ b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.setExecutorsHeld(true)
+ case _ =>
+ }
+ executorAllocationManager match {
+ case Some(manager) => manager.suspend()
+ case None =>
+ b.requestTotalExecutors(
+
immutable.Map(resourceProfileManager.defaultResourceProfile.id -> 0),
Review Comment:
Fixed in c577f4d as suggested: `reset()` re-asserts the zero requirement for
the default profile when held and calls `doRequestTotalExecutors` without
awaiting.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && 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 through the
external shuffle
+ * service, so the application can later pick up where it left off via
`resumeExecutors()`.
+ *
+ * This requires both an external shuffle service
(`spark.shuffle.service.enabled`) and
+ * decommission support (`spark.decommission.enabled`).
+ *
+ * @return whether the request is received by the cluster manager.
Review Comment:
Fixed in c577f4d. Both methods now return the actual ack, with the semantics
documented in the scaladoc: hold drains regardless (with dynamic allocation the
rejected push is retried in the background); resume lifts the hold when dynamic
allocation owns the retry, and otherwise keeps the executors held on a rejected
push so the call can be retried.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && conf.get(DECOMMISSION_ENABLED)
Review Comment:
Fixed in c577f4d. `executorHoldSupported` now accepts
`shuffleDriverComponents.supportsReliableStorage()` as an alternative to the
external shuffle service, and the `require` message and docs mention both.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && 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 through the
external shuffle
+ * service, so the application can later pick up where it left off via
`resumeExecutors()`.
Review Comment:
Added in c577f4d to the scaladoc, the `spark.ui.holdEnabled` doc, and
`configuration.md`.
--
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]
