[
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, and even with a fixed number of
executors the spec cannot tell how many are running or whether they have
started. The {{SparkApplication}} status does not declare its executors today,
so anything that needs them, such as users, dashboards, autoscalers and batch
schedulers, has to list and watch the executor pods itself and know which pods
belong to the current attempt.
This gap blocks Kueue today. Kueue's chosen approach to elastic Spark
applications is to follow executor counts and states that the operator declares
in the status (kubernetes-sigs/kueue#17127:
https://github.com/kubernetes-sigs/kueue/pull/17127). The Kueue community
maintains a built-in framework integration for the Kubeflow Spark Operator, and
that solution supports only that operator. Declaring the same information here
closes the gap for this operator's applications, and gives every other consumer
the same answer from the {{SparkApplication}} alone.
This proposes four new optional fields in {{SparkApplication}} status, for the
current attempt:
||Field||Recorded for||Meaning||Lets a consumer||
|{{status.activeExecutors}}|every application|executor pods that are pending
(including scheduling-gated) or running (including being deleted)|see the
executors an application holds now, e.g. to size it|
|{{status.readyExecutors}}|every application|executor pods that are running and
ready, and not being deleted|tell whether the executors have started, e.g. for
a readiness timeout|
|{{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. how many executors the driver has requested
in total|
|{{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.
* The two counts can be turned off with
{{spark.kubernetes.operator.reconciler.executorCountsEnabled}} (on by default,
dynamically overridable), in case their status writes load the API server and
etcd more than a cluster can take. Turning it off clears the counts recorded
before; {{maxExecutorId}} and {{executorState}} are not affected.
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 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.
> 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, and even with a fixed number of
> executors the spec cannot tell how many are running or whether they have
> started. The {{SparkApplication}} status does not declare its executors
> today, so anything that needs them, such as users, dashboards, autoscalers
> and batch schedulers, has to list and watch the executor pods itself and know
> which pods belong to the current attempt.
> This gap blocks Kueue today. Kueue's chosen approach to elastic Spark
> applications is to follow executor counts and states that the operator
> declares in the status (kubernetes-sigs/kueue#17127:
> https://github.com/kubernetes-sigs/kueue/pull/17127). The Kueue community
> maintains a built-in framework integration for the Kubeflow Spark Operator,
> and that solution supports only that operator. Declaring the same information
> here closes the gap for this operator's applications, and gives every other
> consumer the same answer from the {{SparkApplication}} alone.
> This proposes four new optional fields in {{SparkApplication}} status, for
> the current attempt:
> ||Field||Recorded for||Meaning||Lets a consumer||
> |{{status.activeExecutors}}|every application|executor pods that are pending
> (including scheduling-gated) or running (including being deleted)|see the
> executors an application holds now, e.g. to size it|
> |{{status.readyExecutors}}|every application|executor pods that are running
> and ready, and not being deleted|tell whether the executors have started,
> e.g. for a readiness timeout|
> |{{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. how many executors the driver
> has requested in total|
> |{{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.
> * The two counts can be turned off with
> {{spark.kubernetes.operator.reconciler.executorCountsEnabled}} (on by
> default, dynamically overridable), in case their status writes load the API
> server and etcd more than a cluster can take. Turning it off clears the
> counts recorded before; {{maxExecutorId}} and {{executorState}} are not
> affected.
> 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]
