[
https://issues.apache.org/jira/browse/SPARK-61258?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Purushotham Pushpavanthar updated SPARK-61258:
----------------------------------------------
Description:
h3. SPIP: Ratio-constrained heterogeneous executors on Kubernetes
Full SPIP document:
https://docs.google.com/document/d/1PCTHXYIgzDQLmNO9KUjBxoAkG36z-P7J_t9SDvaPjzI/edit
h4. Q1. What are you trying to do?
Let a Spark job on Kubernetes run executors of different sizes as long as every
executor keeps the same memory per CPU core. The user states a ladder of
executor sizes ({{spark.kubernetes.executor.shape.ladder=16,8,4}}) instead of
one fixed size; heap per core is derived from spark.executor.memory /
spark.executor.cores. The driver requests the largest shape first; when
Kubernetes reports the pod Unschedulable past a timeout, the driver replaces it
with pods one rung down that sum to the same slot count, and periodically
probes upward again. Nothing visible to the task scheduler changes:
{{spark.task.cpus}}, memory per core and custom resources per task stay
uniform; only cores per executor and total memory per executor float.
Everything outside the Kubernetes allocator keeps counting executors in
reference-shape units, so {{spark.dynamicAllocation.*}} and
{{SparkContext.requestTotalExecutors}} keep their meaning.
h4. Q2. What problem is this NOT designed to solve?
* Different memory-per-core ratios inside one ResourceProfile (use two
profiles).
* Growing or shrinking a running executor (heap and registered cores are fixed
at start).
* Choosing node types, zones or spot vs on-demand. That stays with the cluster
(GKE compute classes, Karpenter NodePools, cluster-autoscaler).
* Reading node capacity from the driver (optional later strategy, needs node
RBAC).
* YARN / Standalone. Core model is cluster-manager neutral; only the Kubernetes
{{direct}} pods allocator is implemented. StatefulSet and Deployment allocators
reject the config.
h4. Q3. How is it done today, and what are the limits?
Homogeneity is an accident of the allocation path, not a scheduler requirement.
{{ExecutorAllocationManager}} targets an executor count per profile;
{{ExecutorPodsAllocator}} reconciles counts per rpId and never looks at pod
size or a Pending pod's {{PodScheduled=False}} condition;
{{BasicExecutorFeatureStep}} derives pod shape purely from the profile.
{{RegisterExecutor}} already carries per-executor cores, so the task scheduler
tolerates heterogeneous cores today. {{ResourceProfileManager.canBeScheduled}}
requires exact rpId match, so sibling profiles per size would not work. Users
work around it with small homogeneous executors and pay per-JVM overhead (384
MiB overhead floor, smaller unified memory pools, N x M shuffle connections,
broadcast per JVM, slower spot decommission).
h4. Q4. What is new and why will it succeed?
* The invariant is memory per core, not pod size, so this is an allocator and
profile change, not a scheduler change.
* One ResourceProfile carries a ladder of shapes under one rpId;
{{canBeScheduled}} is unchanged. Phase 1 is configuration-only: no change to
the ResourceProfile API or the scheduler backend contract; the Kubernetes
backend converts the executor-count target to cores itself.
* Kubernetes tells the driver what fits through the pod watch Spark already has
(new {{PodUnschedulable}} state); no new RBAC.
* On clusters with compute classes the two decisions compose: the platform
picks node type and capacity type, Spark picks pod size. Node-aware packing in
the driver (approach B) would duplicate the platform's job, needs
cluster-scoped node RBAC and fights consolidation.
h4. Q5. Who cares?
Platform teams running Spark on mixed or spot fleets (higher node fill, higher
spot obtainability), job owners (one ratio replaces per-cluster size tuning,
per-task memory stays comparable), and operators of autoscaled clusters (the
feature cooperates with the autoscaler instead of bypassing it).
h4. Q6. Risks
Downshift fires only on an insufficient cpu/memory scheduler message, not on
taint, affinity or PVC causes; the upshift probe backs off exponentially and
carries a low PriorityClass so it never triggers an autoscaler scale-up.
Downshift timeout vs autoscaler provisioning latency (mitigated by reading
{{TriggeredScaleUp}} / {{NotTriggerScaleUp}} / {{Nominated}} pod events); pod
churn on downshift (bounded by {{maxPendingPods}}, reuses the excess-pending
deletion path); executor-side conf readers seeing app-level
{{spark.executor.memory}} / {{cores}} (executor backend writes pod values into
its local SparkConf); overhead floor squeezing small shapes (per-shape split
and warning); idle removal under-shooting the slot target (subtract real
slots); UI and history server (additive event fields).
h4. Q7. How long will it take?
About two months for Phase 1 across four PRs, all inside the Kubernetes module
plus one backend method: ladder allocation and per-rung sizing (prototyped);
scheduler backend accounting (registered-cores gate, kill-path weights,
executor cores from the launch argument, metrics); upshift probe and autoscaler
events; docs. Prerequisites SPARK-61267 and SPARK-61268 are already open. The
ResourceProfile API for ladders on custom profiles is Phase 2 and not part of
this vote.
h4. Q8. Exams
Mid-term: a single-rung ladder behaves byte-for-byte like today in
{{TaskSchedulerImplSuite}} and {{ExecutorAllocationManagerSuite}}; executors
report pod values in SparkEnv.conf and the Executors tab. Final:
{{ExecutorPodsAllocatorSuite}} downshift and upshift scenarios; k8s integration
test on a two-node-size cluster where total slots match the target and every
executor has memory / cores equal to the ratio; StatefulSet and Deployment
allocators fail fast.
Design baseline: apache/spark master, commit 4686647b3a1 (2026-11-09).
was:
h3. SPIP: Ratio-constrained heterogeneous executors on Kubernetes
Full SPIP document:
https://docs.google.com/document/d/1_bnwJTVVWUuK0vC3yhZAJ4d7uI3Qp8sjGGdIn8Geld8/edit
h4. Q1. What are you trying to do?
Let a Spark job on Kubernetes run executors of different sizes as long as every
executor keeps the same memory per CPU core. The user states a ladder of
executor sizes ({{spark.kubernetes.executor.shape.ladder=16,8,4}}) instead of
one fixed size; heap per core is derived from spark.executor.memory /
spark.executor.cores. The driver requests the largest shape first; when
Kubernetes reports the pod Unschedulable past a timeout, the driver replaces it
with pods one rung down that sum to the same slot count, and periodically
probes upward again. Nothing visible to the task scheduler changes:
{{spark.task.cpus}}, memory per core and custom resources per task stay
uniform; only cores per executor and total memory per executor float.
Everything outside the Kubernetes allocator keeps counting executors in
reference-shape units, so {{spark.dynamicAllocation.*}} and
{{SparkContext.requestTotalExecutors}} keep their meaning.
h4. Q2. What problem is this NOT designed to solve?
* Different memory-per-core ratios inside one ResourceProfile (use two
profiles).
* Growing or shrinking a running executor (heap and registered cores are fixed
at start).
* Choosing node types, zones or spot vs on-demand. That stays with the cluster
(GKE compute classes, Karpenter NodePools, cluster-autoscaler).
* Reading node capacity from the driver (optional later strategy, needs node
RBAC).
* YARN / Standalone. Core model is cluster-manager neutral; only the Kubernetes
{{direct}} pods allocator is implemented. StatefulSet and Deployment allocators
reject the config.
h4. Q3. How is it done today, and what are the limits?
Homogeneity is an accident of the allocation path, not a scheduler requirement.
{{ExecutorAllocationManager}} targets an executor count per profile;
{{ExecutorPodsAllocator}} reconciles counts per rpId and never looks at pod
size or a Pending pod's {{PodScheduled=False}} condition;
{{BasicExecutorFeatureStep}} derives pod shape purely from the profile.
{{RegisterExecutor}} already carries per-executor cores, so the task scheduler
tolerates heterogeneous cores today. {{ResourceProfileManager.canBeScheduled}}
requires exact rpId match, so sibling profiles per size would not work. Users
work around it with small homogeneous executors and pay per-JVM overhead (384
MiB overhead floor, smaller unified memory pools, N x M shuffle connections,
broadcast per JVM, slower spot decommission).
h4. Q4. What is new and why will it succeed?
* The invariant is memory per core, not pod size, so this is an allocator and
profile change, not a scheduler change.
* One ResourceProfile carries a ladder of shapes under one rpId;
{{canBeScheduled}} is unchanged. Phase 1 is configuration-only: no change to
the ResourceProfile API or the scheduler backend contract; the Kubernetes
backend converts the executor-count target to cores itself.
* Kubernetes tells the driver what fits through the pod watch Spark already has
(new {{PodUnschedulable}} state); no new RBAC.
* On clusters with compute classes the two decisions compose: the platform
picks node type and capacity type, Spark picks pod size. Node-aware packing in
the driver (approach B) would duplicate the platform's job, needs
cluster-scoped node RBAC and fights consolidation.
h4. Q5. Who cares?
Platform teams running Spark on mixed or spot fleets (higher node fill, higher
spot obtainability), job owners (one ratio replaces per-cluster size tuning,
per-task memory stays comparable), and operators of autoscaled clusters (the
feature cooperates with the autoscaler instead of bypassing it).
h4. Q6. Risks
Downshift fires only on an insufficient cpu/memory scheduler message, not on
taint, affinity or PVC causes; the upshift probe backs off exponentially and
carries a low PriorityClass so it never triggers an autoscaler scale-up.
Downshift timeout vs autoscaler provisioning latency (mitigated by reading
{{TriggeredScaleUp}} / {{NotTriggerScaleUp}} / {{Nominated}} pod events); pod
churn on downshift (bounded by {{maxPendingPods}}, reuses the excess-pending
deletion path); executor-side conf readers seeing app-level
{{spark.executor.memory}} / {{cores}} (executor backend writes pod values into
its local SparkConf); overhead floor squeezing small shapes (per-shape split
and warning); idle removal under-shooting the slot target (subtract real
slots); UI and history server (additive event fields).
h4. Q7. How long will it take?
About two months for Phase 1 across four PRs, all inside the Kubernetes module
plus one backend method: ladder allocation and per-rung sizing (prototyped);
scheduler backend accounting (registered-cores gate, kill-path weights,
executor cores from the launch argument, metrics); upshift probe and autoscaler
events; docs. Prerequisites SPARK-61267 and SPARK-61268 are already open. The
ResourceProfile API for ladders on custom profiles is Phase 2 and not part of
this vote.
h4. Q8. Exams
Mid-term: a single-rung ladder behaves byte-for-byte like today in
{{TaskSchedulerImplSuite}} and {{ExecutorAllocationManagerSuite}}; executors
report pod values in SparkEnv.conf and the Executors tab. Final:
{{ExecutorPodsAllocatorSuite}} downshift and upshift scenarios; k8s integration
test on a two-node-size cluster where total slots match the target and every
executor has memory / cores equal to the ratio; StatefulSet and Deployment
allocators fail fast.
Design baseline: apache/spark master, commit 4686647b3a1 (2026-11-09).
> SPIP: Ratio-constrained heterogeneous executors on Kubernetes
> -------------------------------------------------------------
>
> Key: SPARK-61258
> URL: https://issues.apache.org/jira/browse/SPARK-61258
> Project: Spark
> Issue Type: New Feature
> Components: Kubernetes
> Affects Versions: 5.0.0
> Reporter: Purushotham Pushpavanthar
> Priority: Major
>
> h3. SPIP: Ratio-constrained heterogeneous executors on Kubernetes
> Full SPIP document:
> https://docs.google.com/document/d/1PCTHXYIgzDQLmNO9KUjBxoAkG36z-P7J_t9SDvaPjzI/edit
> h4. Q1. What are you trying to do?
> Let a Spark job on Kubernetes run executors of different sizes as long as
> every executor keeps the same memory per CPU core. The user states a ladder
> of executor sizes ({{spark.kubernetes.executor.shape.ladder=16,8,4}}) instead
> of one fixed size; heap per core is derived from spark.executor.memory /
> spark.executor.cores. The driver requests the largest shape first; when
> Kubernetes reports the pod Unschedulable past a timeout, the driver replaces
> it with pods one rung down that sum to the same slot count, and periodically
> probes upward again. Nothing visible to the task scheduler changes:
> {{spark.task.cpus}}, memory per core and custom resources per task stay
> uniform; only cores per executor and total memory per executor float.
> Everything outside the Kubernetes allocator keeps counting executors in
> reference-shape units, so {{spark.dynamicAllocation.*}} and
> {{SparkContext.requestTotalExecutors}} keep their meaning.
> h4. Q2. What problem is this NOT designed to solve?
> * Different memory-per-core ratios inside one ResourceProfile (use two
> profiles).
> * Growing or shrinking a running executor (heap and registered cores are
> fixed at start).
> * Choosing node types, zones or spot vs on-demand. That stays with the
> cluster (GKE compute classes, Karpenter NodePools, cluster-autoscaler).
> * Reading node capacity from the driver (optional later strategy, needs node
> RBAC).
> * YARN / Standalone. Core model is cluster-manager neutral; only the
> Kubernetes {{direct}} pods allocator is implemented. StatefulSet and
> Deployment allocators reject the config.
> h4. Q3. How is it done today, and what are the limits?
> Homogeneity is an accident of the allocation path, not a scheduler
> requirement. {{ExecutorAllocationManager}} targets an executor count per
> profile; {{ExecutorPodsAllocator}} reconciles counts per rpId and never looks
> at pod size or a Pending pod's {{PodScheduled=False}} condition;
> {{BasicExecutorFeatureStep}} derives pod shape purely from the profile.
> {{RegisterExecutor}} already carries per-executor cores, so the task
> scheduler tolerates heterogeneous cores today.
> {{ResourceProfileManager.canBeScheduled}} requires exact rpId match, so
> sibling profiles per size would not work. Users work around it with small
> homogeneous executors and pay per-JVM overhead (384 MiB overhead floor,
> smaller unified memory pools, N x M shuffle connections, broadcast per JVM,
> slower spot decommission).
> h4. Q4. What is new and why will it succeed?
> * The invariant is memory per core, not pod size, so this is an allocator and
> profile change, not a scheduler change.
> * One ResourceProfile carries a ladder of shapes under one rpId;
> {{canBeScheduled}} is unchanged. Phase 1 is configuration-only: no change to
> the ResourceProfile API or the scheduler backend contract; the Kubernetes
> backend converts the executor-count target to cores itself.
> * Kubernetes tells the driver what fits through the pod watch Spark already
> has (new {{PodUnschedulable}} state); no new RBAC.
> * On clusters with compute classes the two decisions compose: the platform
> picks node type and capacity type, Spark picks pod size. Node-aware packing
> in the driver (approach B) would duplicate the platform's job, needs
> cluster-scoped node RBAC and fights consolidation.
> h4. Q5. Who cares?
> Platform teams running Spark on mixed or spot fleets (higher node fill,
> higher spot obtainability), job owners (one ratio replaces per-cluster size
> tuning, per-task memory stays comparable), and operators of autoscaled
> clusters (the feature cooperates with the autoscaler instead of bypassing it).
> h4. Q6. Risks
> Downshift fires only on an insufficient cpu/memory scheduler message, not on
> taint, affinity or PVC causes; the upshift probe backs off exponentially and
> carries a low PriorityClass so it never triggers an autoscaler scale-up.
> Downshift timeout vs autoscaler provisioning latency (mitigated by reading
> {{TriggeredScaleUp}} / {{NotTriggerScaleUp}} / {{Nominated}} pod events); pod
> churn on downshift (bounded by {{maxPendingPods}}, reuses the excess-pending
> deletion path); executor-side conf readers seeing app-level
> {{spark.executor.memory}} / {{cores}} (executor backend writes pod values
> into its local SparkConf); overhead floor squeezing small shapes (per-shape
> split and warning); idle removal under-shooting the slot target (subtract
> real slots); UI and history server (additive event fields).
> h4. Q7. How long will it take?
> About two months for Phase 1 across four PRs, all inside the Kubernetes
> module plus one backend method: ladder allocation and per-rung sizing
> (prototyped); scheduler backend accounting (registered-cores gate, kill-path
> weights, executor cores from the launch argument, metrics); upshift probe and
> autoscaler events; docs. Prerequisites SPARK-61267 and SPARK-61268 are
> already open. The ResourceProfile API for ladders on custom profiles is Phase
> 2 and not part of this vote.
> h4. Q8. Exams
> Mid-term: a single-rung ladder behaves byte-for-byte like today in
> {{TaskSchedulerImplSuite}} and {{ExecutorAllocationManagerSuite}}; executors
> report pod values in SparkEnv.conf and the Executors tab. Final:
> {{ExecutorPodsAllocatorSuite}} downshift and upshift scenarios; k8s
> integration test on a two-node-size cluster where total slots match the
> target and every executor has memory / cores equal to the ratio; StatefulSet
> and Deployment allocators fail fast.
> Design baseline: apache/spark master, commit 4686647b3a1 (2026-11-09).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
