[
https://issues.apache.org/jira/browse/SPARK-61258?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141505#comment-18141505
]
Purushotham Pushpavanthar commented on SPARK-61258:
---------------------------------------------------
Prototype run of the shape ladder (branch
[SPARK-61258-ladder-prototype|https://github.com/pushpavanthar/spark/tree/SPARK-61258-ladder-prototype],
results and scripts under {{dev/spip-61258-ladder-demo/}}).
*Setup:* KWOK fake cluster with one 16-cpu/64Gi node and two 8-cpu/32Gi nodes.
Driver: {{spark.executor.cores=4}}, {{spark.executor.memory=4g}},
{{spark.executor.instances=8}} (32 cores wanted),
{{spark.kubernetes.executor.shape.ladder=16,8,4}}, {{downshiftTimeout=30s}}.
*Observed:*
* t+0s: exec-1 (16 cores, 18022Mi) Running on big-16cpu; exec-2 (16 cores)
Pending, scheduler message {{0/3 nodes are available: 3 Insufficient cpu}}.
* t+30s: allocator logs "moving the shape ladder down to 8 cores per executor",
deletes exec-2 and requests two 8-core pods.
* t+51s: exec-3 and exec-4 (8 cores, 9011Mi each) Running on small-8cpu-a and
small-8cpu-b.
* Final fill 16 + 8 + 8 = 32 cores = target; heap per core constant (1g/core
plus 10% overhead on every rung).
A homogeneous 16-core setting would strand both 8-cpu nodes; a homogeneous
4-core setting reaches the target with 8 JVMs instead of 3. Prototype scope:
default profile only, downshift only (no upshift probe yet), no excess cleanup
under a ladder.
> 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]
