[
https://issues.apache.org/jira/browse/SPARK-61130?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Pradeep Reddy updated SPARK-61130:
----------------------------------
Description:
Spark Dynamic Allocation adds and removes executors from the driver without
changing the {{SparkApplication}} spec, so a scheduler cannot tell from the
spec how many executors an application holds or whether they have started. This
enhancement will allow External Framework Integrations of Apache Spark Operator
for Kueue to support Spark Dynamic Allocation with Kueue's
ElasticJobsViaWorkloadSlices using the CR declared executor counts. The Kueue
community maintains a built-in framework integration for the Kubeflow Spark
Operator, and implemented Spark Dynamic Allocation support part of
(kubernetes-sigs/kueue#17127:
https://github.com/kubernetes-sigs/kueue/pull/17127) only supports that
operator.
This proposes four new optional fields in {{SparkApplication}} status, for the
current attempt:
||Field||Recorded for||Meaning||Used by a scheduler to||
|{{status.activeExecutors}}|every application|executor pods that are pending
(including scheduling-gated) or running (including being deleted)|size the
application: the executors it holds now|
|{{status.readyExecutors}}|every application|executor pods that are running and
ready, and not being deleted|tell whether the executors have started, e.g. for
Kueue's {{waitForPodsReady}}|
|{{status.maxExecutorId}}|applications with Dynamic Allocation|highest executor
ID so far; Spark never reuses executor IDs, so it never decreases within an
attempt|tell scale-ups apart, e.g. to give each Kueue Workload slice a unique
name|
|{{status.executorState}}|applications with Dynamic Allocation, opt-in (off by
default)|executor pod name -> {{PENDING}}, {{RUNNING}}, {{COMPLETED}},
{{FAILED}} or {{UNKNOWN}}|follow each executor, when the counts are not enough|
h3. Executor counts and highest executor ID
* Computed from the executor pods of the current attempt
({{spark-app-selector}} equal to the attempt's {{spark.app.id}}) in the
existing pod informer cache: no new watch, LIST or GET. They do not depend on
{{executorState}}, so they stay exact however many executors an application has
had.
* Unset until the attempt has an executor pod. The counts drop to 0 once every
executor is gone, while {{maxExecutorId}} is kept. All of them start over with
each attempt and when the application is suspended.
h3. Executor states (opt-in)
* Off by default, turned on with
{{spark.kubernetes.operator.reconciler.executorStateEnabled}} (dynamically
overridable). A scheduler that reads only the counts does not need them, and
every entry adds to each status write.
* An executor's state follows its pod phase ({{Pending}}, {{Running}},
{{Succeeded}}, {{Failed}}; {{UNKNOWN}} otherwise), and the entry is kept after
its pod is gone. The informer's delete event records the pod's last phase, so
an executor that completed or failed as Dynamic Allocation removed it is
recorded as {{COMPLETED}} or {{FAILED}} rather than {{UNKNOWN}}. Otherwise an
executor whose pod is gone becomes {{UNKNOWN}} while the driver runs, then
{{COMPLETED}} or {{FAILED}} with the application.
* Bounded by {{spark.kubernetes.operator.reconciler.maxTrackedExecutorPerApp}}
(default 1000, dynamically overridable) to keep the status within the size
limit of a Kubernetes object. Beyond it, executors that are gone are evicted
first, oldest first, so a long-running application keeps recording its new
executors.
h3. API server load
* The status is written only when a recorded field changes, in one patch per
reconciliation, so the writes are bounded by the existing per-application
reconcile rate limiter ({{maxLoopForPeriod}} per {{refreshPeriodSeconds}}).
Several executor changes within a rate-limit window are coalesced into one
write.
* All fields are new and optional, v1 CRD only, so existing users are
unaffected.
was:
Spark Dynamic Allocation adds and removes executors from the driver without
changing the {{SparkApplication}} spec, so a scheduler cannot tell from the
spec how many executors an application holds or whether they have started. This
enhancement will allow External Framework Integrations of Apache Spark Operator
for Kueue to support Spark Dynamic Allocation using
ElasticJobsViaWorkloadSlices using the CR declared executor counts. The Kueue
community maintains a built-in framework integration for the Kubeflow Spark
Operator, and implemented Spark Dynamic Allocation support part of
(kubernetes-sigs/kueue#17127:
https://github.com/kubernetes-sigs/kueue/pull/17127) only supports that
operator.
This proposes four new optional fields in {{SparkApplication}} status, for the
current attempt:
||Field||Recorded for||Meaning||Used by a scheduler to||
|{{status.activeExecutors}}|every application|executor pods that are pending
(including scheduling-gated) or running (including being deleted)|size the
application: the executors it holds now|
|{{status.readyExecutors}}|every application|executor pods that are running and
ready, and not being deleted|tell whether the executors have started, e.g. for
Kueue's {{waitForPodsReady}}|
|{{status.maxExecutorId}}|applications with Dynamic Allocation|highest executor
ID so far; Spark never reuses executor IDs, so it never decreases within an
attempt|tell scale-ups apart, e.g. to give each Kueue Workload slice a unique
name|
|{{status.executorState}}|applications with Dynamic Allocation, opt-in (off by
default)|executor pod name -> {{PENDING}}, {{RUNNING}}, {{COMPLETED}},
{{FAILED}} or {{UNKNOWN}}|follow each executor, when the counts are not enough|
h3. Executor counts and highest executor ID
* Computed from the executor pods of the current attempt
({{spark-app-selector}} equal to the attempt's {{spark.app.id}}) in the
existing pod informer cache: no new watch, LIST or GET. They do not depend on
{{executorState}}, so they stay exact however many executors an application has
had.
* Unset until the attempt has an executor pod. The counts drop to 0 once every
executor is gone, while {{maxExecutorId}} is kept. All of them start over with
each attempt and when the application is suspended.
h3. Executor states (opt-in)
* Off by default, turned on with
{{spark.kubernetes.operator.reconciler.executorStateEnabled}} (dynamically
overridable). A scheduler that reads only the counts does not need them, and
every entry adds to each status write.
* An executor's state follows its pod phase ({{Pending}}, {{Running}},
{{Succeeded}}, {{Failed}}; {{UNKNOWN}} otherwise), and the entry is kept after
its pod is gone. The informer's delete event records the pod's last phase, so
an executor that completed or failed as Dynamic Allocation removed it is
recorded as {{COMPLETED}} or {{FAILED}} rather than {{UNKNOWN}}. Otherwise an
executor whose pod is gone becomes {{UNKNOWN}} while the driver runs, then
{{COMPLETED}} or {{FAILED}} with the application.
* Bounded by {{spark.kubernetes.operator.reconciler.maxTrackedExecutorPerApp}}
(default 1000, dynamically overridable) to keep the status within the size
limit of a Kubernetes object. Beyond it, executors that are gone are evicted
first, oldest first, so a long-running application keeps recording its new
executors.
h3. API server load
* The status is written only when a recorded field changes, in one patch per
reconciliation, so the writes are bounded by the existing per-application
reconcile rate limiter ({{maxLoopForPeriod}} per {{refreshPeriodSeconds}}).
Several executor changes within a rate-limit window are coalesced into one
write.
* All fields are new and optional, v1 CRD only, so existing users are
unaffected.
> Add activeExecutors, readyExecutors, maxExecutorId and ExecutorStates to
> SparkApplication status
> -------------------------------------------------------------------------------------------------
>
> Key: SPARK-61130
> URL: https://issues.apache.org/jira/browse/SPARK-61130
> Project: Spark
> Issue Type: Improvement
> Components: Kubernetes
> Affects Versions: kubernetes-operator-1.1.0
> Reporter: Pradeep Reddy
> Priority: Major
> Fix For: kubernetes-operator-1.1.0
>
>
> Spark Dynamic Allocation adds and removes executors from the driver without
> changing the {{SparkApplication}} spec, so a scheduler cannot tell from the
> spec how many executors an application holds or whether they have started.
> This enhancement will allow External Framework Integrations of Apache Spark
> Operator for Kueue to support Spark Dynamic Allocation with Kueue's
> ElasticJobsViaWorkloadSlices using the CR declared executor counts. The Kueue
> community maintains a built-in framework integration for the Kubeflow Spark
> Operator, and implemented Spark Dynamic Allocation support part of
> (kubernetes-sigs/kueue#17127:
> https://github.com/kubernetes-sigs/kueue/pull/17127) only supports that
> operator.
> This proposes four new optional fields in {{SparkApplication}} status, for
> the current attempt:
> ||Field||Recorded for||Meaning||Used by a scheduler to||
> |{{status.activeExecutors}}|every application|executor pods that are pending
> (including scheduling-gated) or running (including being deleted)|size the
> application: the executors it holds now|
> |{{status.readyExecutors}}|every application|executor pods that are running
> and ready, and not being deleted|tell whether the executors have started,
> e.g. for Kueue's {{waitForPodsReady}}|
> |{{status.maxExecutorId}}|applications with Dynamic Allocation|highest
> executor ID so far; Spark never reuses executor IDs, so it never decreases
> within an attempt|tell scale-ups apart, e.g. to give each Kueue Workload
> slice a unique name|
> |{{status.executorState}}|applications with Dynamic Allocation, opt-in (off
> by default)|executor pod name -> {{PENDING}}, {{RUNNING}}, {{COMPLETED}},
> {{FAILED}} or {{UNKNOWN}}|follow each executor, when the counts are not
> enough|
> h3. Executor counts and highest executor ID
> * Computed from the executor pods of the current attempt
> ({{spark-app-selector}} equal to the attempt's {{spark.app.id}}) in the
> existing pod informer cache: no new watch, LIST or GET. They do not depend on
> {{executorState}}, so they stay exact however many executors an application
> has had.
> * Unset until the attempt has an executor pod. The counts drop to 0 once
> every executor is gone, while {{maxExecutorId}} is kept. All of them start
> over with each attempt and when the application is suspended.
> h3. Executor states (opt-in)
> * Off by default, turned on with
> {{spark.kubernetes.operator.reconciler.executorStateEnabled}} (dynamically
> overridable). A scheduler that reads only the counts does not need them, and
> every entry adds to each status write.
> * An executor's state follows its pod phase ({{Pending}}, {{Running}},
> {{Succeeded}}, {{Failed}}; {{UNKNOWN}} otherwise), and the entry is kept
> after its pod is gone. The informer's delete event records the pod's last
> phase, so an executor that completed or failed as Dynamic Allocation removed
> it is recorded as {{COMPLETED}} or {{FAILED}} rather than {{UNKNOWN}}.
> Otherwise an executor whose pod is gone becomes {{UNKNOWN}} while the driver
> runs, then {{COMPLETED}} or {{FAILED}} with the application.
> * Bounded by
> {{spark.kubernetes.operator.reconciler.maxTrackedExecutorPerApp}} (default
> 1000, dynamically overridable) to keep the status within the size limit of a
> Kubernetes object. Beyond it, executors that are gone are evicted first,
> oldest first, so a long-running application keeps recording its new executors.
> h3. API server load
> * The status is written only when a recorded field changes, in one patch per
> reconciliation, so the writes are bounded by the existing per-application
> reconcile rate limiter ({{maxLoopForPeriod}} per {{refreshPeriodSeconds}}).
> Several executor changes within a rate-limit window are coalesced into one
> write.
> * All fields are new and optional, v1 CRD only, so existing users are
> unaffected.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
