[
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 design doc (comments open):
https://docs.google.com/document/d/1a5zFhMnTz7Tftb7fZgnri58RdmixQcHamEQieva5l5s/edit
h4. Q1. What are you trying to do?
I want a Spark job on Kubernetes to run executors of different sizes, as long
as every executor keeps the same memory per core. Today one job means one pod
shape. On a cluster with mixed node sizes, or where spot capacity comes and
goes, that single shape either strands capacity or pushes everyone down to
small executors.
The idea: instead of one fixed size the user gives a ladder, e.g.
{{spark.kubernetes.executor.shape.ladder=16,8,4}}. Memory per core still comes
from {{spark.executor.memory}} / {{spark.executor.cores}}. The driver asks for
the biggest shape first. If Kubernetes leaves that pod Unschedulable for
insufficient cpu/memory past a timeout, the driver swaps it for pods one rung
down that add up to the same number of cores, and later probes upward again
with backoff.
The task scheduler does not see any of this. {{spark.task.cpus}}, memory per
core and per-task custom resources stay uniform; only cores per executor and
total memory per executor float. Everything outside the k8s allocator keeps
counting executors in units of the reference shape, so
{{spark.dynamicAllocation.*}} and {{SparkContext.requestTotalExecutors}} mean
exactly what they mean today.
h4. Q2. What problem is this NOT designed to solve?
* Different memory/core ratios inside one ResourceProfile. That is what a
second profile is for.
* Resizing a running executor. Heap and registered cores are fixed at launch.
* Picking node types, zones, or spot vs on-demand. I deliberately leave that
with the cluster (GKE compute classes, Karpenter NodePools, cluster-autoscaler).
* Reading node capacity from the driver. Possible later as an optional
strategy, but it needs node RBAC and I do not want that in Phase 1.
* YARN / Standalone. The core model is cluster-manager neutral, but only the
Kubernetes {{direct}} pods allocator gets an implementation. StatefulSet and
Deployment allocators reject the config.
h4. Q3. How is it done today, and what are the limits?
Reading the code, homogeneity is an accident of the allocation path, not
something the scheduler needs. {{ExecutorAllocationManager}} targets an
executor count per profile. {{ExecutorPodsAllocator}} reconciles counts per
rpId and never looks at pod size, nor at a Pending pod's {{PodScheduled=False}}
condition. {{BasicExecutorFeatureStep}} derives the pod shape purely from the
profile. Meanwhile {{RegisterExecutor}} already carries per-executor cores, so
the task scheduler copes with mixed cores right now. What does not work is
faking it with one sibling profile per size, because
{{ResourceProfileManager.canBeScheduled}} needs an exact rpId match.
So people run small homogeneous executors and eat the per-JVM cost: the 384 MiB
overhead floor on each one, smaller unified memory pools, N x M shuffle
connections, one broadcast copy per JVM, slower decommission on spot.
h4. Q4. What is new and why will it succeed?
* The invariant is memory per core, not pod size. That makes it an allocator
and profile change, not a scheduler change.
* One ResourceProfile carries the ladder under one rpId, so {{canBeScheduled}}
is untouched. Phase 1 is config-only: no ResourceProfile API change, no
scheduler backend contract change. The k8s backend converts the executor-count
target to cores on its own.
* Kubernetes tells the driver what fits through the pod watch Spark already
runs (new {{PodUnschedulable}} state). No new RBAC.
* On clusters with compute classes the two decisions compose: the platform
picks node and capacity type, Spark picks pod size. I looked at node-aware
packing in the driver (approach B in the doc) and dropped it. It duplicates 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: better node fill, better
spot obtainability. Job owners: tune one ratio instead of a size per cluster,
and per-task memory stays comparable across clusters. Operators of autoscaled
clusters: this cooperates with the autoscaler instead of going around it.
h4. Q6. Risks
The ones I know about, and what the design does about each:
* Downshift fires only on an insufficient cpu/memory scheduler message, never
on taint, affinity or PVC causes.
* The upshift probe backs off exponentially and runs under a low PriorityClass,
so it cannot trigger an autoscaler scale-up on its own.
* Downshift timeout racing autoscaler provisioning: read the
{{TriggeredScaleUp}} / {{NotTriggerScaleUp}} / {{Nominated}} pod events before
giving up on a rung.
* Pod churn on downshift: bounded by {{maxPendingPods}}, reuses the existing
excess-pending deletion path.
* Executor-side conf readers seeing the app-level {{spark.executor.memory}} /
{{cores}}: the executor backend writes the pod's real values into its local
SparkConf.
* Overhead floor squeezing the small rungs: per-shape split plus a warning.
* Idle removal in {{ExecutorAllocationManager}} counts executors, so under a
ladder it would never release an idle big one. Weighting removal by reference
units is the one core change in Phase 1.
* UI / history server: additive event fields only.
h4. Q7. How long will it take?
Roughly two months for Phase 1, in four PRs: ladder allocation and per-rung
sizing (prototyped, see the comment below); scheduler backend accounting
(registered-cores gate, kill-path weights, executor cores from the launch
argument, metrics) together with the unit-weighted removal in EAM; upshift
probe and autoscaler events; docs. Apart from that one EAM change everything
lives in the Kubernetes module. Prerequisites SPARK-61267 and SPARK-61268 are
already open. A 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}}, and
executors report pod values in SparkEnv.conf and the Executors tab. Final:
{{ExecutorPodsAllocatorSuite}} downshift and upshift scenarios; a k8s
integration test on a cluster with two node sizes where total cores land on
target and every executor has memory/cores equal to the ratio; StatefulSet and
Deployment allocators fail fast.
Design baseline: apache/spark master at 4686647b3a1 (2026-11-09).
was:
h3. SPIP: Ratio-constrained heterogeneous executors on Kubernetes
Full SPIP document:
https://docs.google.com/document/d/1a5zFhMnTz7Tftb7fZgnri58RdmixQcHamEQieva5l5s/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 design doc (comments open):
> https://docs.google.com/document/d/1a5zFhMnTz7Tftb7fZgnri58RdmixQcHamEQieva5l5s/edit
> h4. Q1. What are you trying to do?
> I want a Spark job on Kubernetes to run executors of different sizes, as long
> as every executor keeps the same memory per core. Today one job means one pod
> shape. On a cluster with mixed node sizes, or where spot capacity comes and
> goes, that single shape either strands capacity or pushes everyone down to
> small executors.
> The idea: instead of one fixed size the user gives a ladder, e.g.
> {{spark.kubernetes.executor.shape.ladder=16,8,4}}. Memory per core still
> comes from {{spark.executor.memory}} / {{spark.executor.cores}}. The driver
> asks for the biggest shape first. If Kubernetes leaves that pod Unschedulable
> for insufficient cpu/memory past a timeout, the driver swaps it for pods one
> rung down that add up to the same number of cores, and later probes upward
> again with backoff.
> The task scheduler does not see any of this. {{spark.task.cpus}}, memory per
> core and per-task custom resources stay uniform; only cores per executor and
> total memory per executor float. Everything outside the k8s allocator keeps
> counting executors in units of the reference shape, so
> {{spark.dynamicAllocation.*}} and {{SparkContext.requestTotalExecutors}} mean
> exactly what they mean today.
> h4. Q2. What problem is this NOT designed to solve?
> * Different memory/core ratios inside one ResourceProfile. That is what a
> second profile is for.
> * Resizing a running executor. Heap and registered cores are fixed at launch.
> * Picking node types, zones, or spot vs on-demand. I deliberately leave that
> with the cluster (GKE compute classes, Karpenter NodePools,
> cluster-autoscaler).
> * Reading node capacity from the driver. Possible later as an optional
> strategy, but it needs node RBAC and I do not want that in Phase 1.
> * YARN / Standalone. The core model is cluster-manager neutral, but only the
> Kubernetes {{direct}} pods allocator gets an implementation. StatefulSet and
> Deployment allocators reject the config.
> h4. Q3. How is it done today, and what are the limits?
> Reading the code, homogeneity is an accident of the allocation path, not
> something the scheduler needs. {{ExecutorAllocationManager}} targets an
> executor count per profile. {{ExecutorPodsAllocator}} reconciles counts per
> rpId and never looks at pod size, nor at a Pending pod's
> {{PodScheduled=False}} condition. {{BasicExecutorFeatureStep}} derives the
> pod shape purely from the profile. Meanwhile {{RegisterExecutor}} already
> carries per-executor cores, so the task scheduler copes with mixed cores
> right now. What does not work is faking it with one sibling profile per size,
> because {{ResourceProfileManager.canBeScheduled}} needs an exact rpId match.
> So people run small homogeneous executors and eat the per-JVM cost: the 384
> MiB overhead floor on each one, smaller unified memory pools, N x M shuffle
> connections, one broadcast copy per JVM, slower decommission on spot.
> h4. Q4. What is new and why will it succeed?
> * The invariant is memory per core, not pod size. That makes it an allocator
> and profile change, not a scheduler change.
> * One ResourceProfile carries the ladder under one rpId, so
> {{canBeScheduled}} is untouched. Phase 1 is config-only: no ResourceProfile
> API change, no scheduler backend contract change. The k8s backend converts
> the executor-count target to cores on its own.
> * Kubernetes tells the driver what fits through the pod watch Spark already
> runs (new {{PodUnschedulable}} state). No new RBAC.
> * On clusters with compute classes the two decisions compose: the platform
> picks node and capacity type, Spark picks pod size. I looked at node-aware
> packing in the driver (approach B in the doc) and dropped it. It duplicates
> 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: better node fill,
> better spot obtainability. Job owners: tune one ratio instead of a size per
> cluster, and per-task memory stays comparable across clusters. Operators of
> autoscaled clusters: this cooperates with the autoscaler instead of going
> around it.
> h4. Q6. Risks
> The ones I know about, and what the design does about each:
> * Downshift fires only on an insufficient cpu/memory scheduler message, never
> on taint, affinity or PVC causes.
> * The upshift probe backs off exponentially and runs under a low
> PriorityClass, so it cannot trigger an autoscaler scale-up on its own.
> * Downshift timeout racing autoscaler provisioning: read the
> {{TriggeredScaleUp}} / {{NotTriggerScaleUp}} / {{Nominated}} pod events
> before giving up on a rung.
> * Pod churn on downshift: bounded by {{maxPendingPods}}, reuses the existing
> excess-pending deletion path.
> * Executor-side conf readers seeing the app-level {{spark.executor.memory}} /
> {{cores}}: the executor backend writes the pod's real values into its local
> SparkConf.
> * Overhead floor squeezing the small rungs: per-shape split plus a warning.
> * Idle removal in {{ExecutorAllocationManager}} counts executors, so under a
> ladder it would never release an idle big one. Weighting removal by reference
> units is the one core change in Phase 1.
> * UI / history server: additive event fields only.
> h4. Q7. How long will it take?
> Roughly two months for Phase 1, in four PRs: ladder allocation and per-rung
> sizing (prototyped, see the comment below); scheduler backend accounting
> (registered-cores gate, kill-path weights, executor cores from the launch
> argument, metrics) together with the unit-weighted removal in EAM; upshift
> probe and autoscaler events; docs. Apart from that one EAM change everything
> lives in the Kubernetes module. Prerequisites SPARK-61267 and SPARK-61268 are
> already open. A 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}}, and
> executors report pod values in SparkEnv.conf and the Executors tab. Final:
> {{ExecutorPodsAllocatorSuite}} downshift and upshift scenarios; a k8s
> integration test on a cluster with two node sizes where total cores land on
> target and every executor has memory/cores equal to the ratio; StatefulSet
> and Deployment allocators fail fast.
> Design baseline: apache/spark master at 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]
