[
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/1dSYUubR6yLNnSsZiE5Ac3q71I4B0EkgaQ5u5ePbywLw/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 ratio and a
range ({{spark.executor.memoryPerCore=4g}}, executors between 2 and 16 cores)
instead of one fixed size. 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.
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 shape family (ladder of concrete shapes plus a
reference shape) under one rpId; {{canBeScheduled}} is unchanged.
* 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 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 three months for Phase 1 across six independently mergeable PRs: shape
families in ResourceProfile; slot-based allocation target; executor sees pod
truth; k8s pod state and shape plumbing; ladder allocation with downshift and
upshift; docs. Node-aware strategy and cpu-only in-place growth are later
phases, not committed here.
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/1HZUl4pXKTV8guFuNyHaGQmp8BkSncDXLwUzA-GeXRFM/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 ratio and a
range ({{spark.executor.memoryPerCore=4g}}, executors between 2 and 16 cores)
instead of one fixed size. 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.
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 shape family (ladder of concrete shapes plus a
reference shape) under one rpId; {{canBeScheduled}} is unchanged.
* 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 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 three months for Phase 1 across six independently mergeable PRs: shape
families in ResourceProfile; slot-based allocation target; executor sees pod
truth; k8s pod state and shape plumbing; ladder allocation with downshift and
upshift; docs. Node-aware strategy and cpu-only in-place growth are later
phases, not committed here.
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/1dSYUubR6yLNnSsZiE5Ac3q71I4B0EkgaQ5u5ePbywLw/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 ratio
> and a range ({{spark.executor.memoryPerCore=4g}}, executors between 2 and 16
> cores) instead of one fixed size. 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.
> 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 shape family (ladder of concrete shapes plus
> a reference shape) under one rpId; {{canBeScheduled}} is unchanged.
> * 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 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 three months for Phase 1 across six independently mergeable PRs: shape
> families in ResourceProfile; slot-based allocation target; executor sees pod
> truth; k8s pod state and shape plumbing; ladder allocation with downshift and
> upshift; docs. Node-aware strategy and cpu-only in-place growth are later
> phases, not committed here.
> 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]
