[SPARK-43086][CORE] Add configurable bin-packing task placement

[starcatmeow]

What changes were proposed in this pull request?

This PR adds an opt-in bin-packing task placement strategy to TaskSchedulerImpl.

It introduces spark.scheduler.taskPlacement.strategy with two values:

• SPREAD, the default, preserves the existing behavior of cycling through shuffled
  executor offers one task at a time.
• BIN_PACK fills an eligible executor before moving to the next one.

BIN_PACK applies only during the NO_PREF and ANY scheduling passes.
PROCESS_LOCAL, NODE_LOCAL, and RACK_LOCAL retain Spark's existing shuffled offer
order, so locality continues to take precedence over packing.

For bin-packed passes, executors with running tasks or tasks assigned earlier in the
current resourceOffers call are considered before idle executors. Each group is ordered
lexicographically by executor ID. The scheduler sorts offer indices once per
resourceOffers call, leaving shuffledOffers, availableCpus, availableResources,
and the output task buffers in their existing shared index space. It then performs a
stable busy/idle partition for each TaskSet and bin-packed pass so that later TaskSets
observe assignments made earlier in the scheduling round.

The PR also documents the new configuration and adds tests for the configuration,
placement order, locality, multiple TaskSets, custom resources, and barrier tasks.

Why are the changes needed?

SPARK-43086 describes a resource
efficiency issue when dynamic allocation is used with the existing spread placement
behavior.

Dynamic allocation removes an executor only after it becomes idle. When a stage has
fewer tasks than the available cluster slots, spreading those tasks across executors can
keep more executors busy than necessary and delay scale-down.

For example, if four executors have two task slots each, a four-task stage can place one
task on each of four executors with SPREAD. With BIN_PACK, it can instead place two
tasks on each of two executors, allowing the other two executors to become idle and be
removed.

The new strategy is opt-in to avoid changing Spark's existing default placement and
latency behavior.

Does this PR introduce any user-facing change?

Yes. It adds the following configuration:

spark.scheduler.taskPlacement.strategy=BIN_PACK

Supported values are:

• SPREAD: preserves the existing placement behavior and remains the default.
• BIN_PACK: fills busy executors first and then idle executors, using lexicographic
  executor ID order within each group, during the NO_PREF and ANY passes.

Applications that do not set this configuration retain the existing behavior.
Locality-specific placement for PROCESS_LOCAL, NODE_LOCAL, and RACK_LOCAL is
unchanged.

How was this patch tested?

The scheduler unit suite passed:

build/sbt 'core/testOnly org.apache.spark.scheduler.TaskSchedulerImplSuite'

All 122 tests passed. The added coverage includes:

• the default SPREAD behavior;
• configuration parsing and invalid values;
• deterministic BIN_PACK executor ordering;
• running and current-round assigned executors before idle executors;
• re-evaluating busy state between TaskSets;
• preservation of PROCESS_LOCAL and NODE_LOCAL placement;
• bin packing at ANY locality;
• custom resource limits;
• barrier tasks and provisional barrier assignments.

Scalastyle and a full package build also passed:

build/sbt core/scalastyle
build/sbt package

The two strategies were additionally tested with real executor JVMs using:

• local-cluster[4,2,1024];
• four initial and maximum executors;
• two cores per executor;
• dynamic allocation and shuffle tracking enabled;
• one minimum executor;
• a four-second executor idle timeout;
• sequential stages containing 8, 4, and 2 NO_PREF tasks;
• twelve-second tasks, allowing idle executors to be removed while a stage was running.

The steady executor counts observed during each stage were:

Strategy	8-task stage	4-task stage	2-task stage
SPREAD	4	4	2
BIN_PACK	4	2	1

The Spark Event Timeline showed that SPREAD first removed executors during the
two-task stage, while BIN_PACK removed two executors during the four-task stage and
another executor during the two-task stage.

SPREAD:

BIN_PACK:

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)

[sunchao]

LGTM

[sunchao]

cc @viirya @cloud-fan @dongjoon-hyun @peter-toth

[viirya]

I traced the scheduler paths independently and this holds up — including a few spots worth spelling out since it's a placement change.

• SPREAD is behavior-preserving. With useBinPacking = false the new inner do { ... } while (useBinPacking && launchedTask) runs the body exactly once per offer index, so resourceOfferSingleTaskSet still does a single pass launching at most one task per offer, and the outer do/while(launchedTaskAtCurrentMaxLocality) still drives the round-robin. Default behavior is unchanged.
• The busy/idle predicate is well chosen. isExecutorBusy(execId) || availableCpus(i) < shuffledOffers(i).cores — the first term catches executors made busy by prior rounds or earlier addRunningTasks in this round, and the second catches provisional barrier assignments in the current round (which decrement availableCpus but aren't registered as running until the whole barrier set launches). The comment captures this well, and the "treats barrier locality assignments as busy" test pins it.
• Shared index space stays intact. Only the iteration order is reordered via offerIndicesByExecutorId; shuffledOffers / availableCpus / availableResources / tasks are all still addressed by the original index i, so there's no re-indexing hazard, and the sort happens once per resourceOffers call.
• Delay-scheduling accounting is unaffected. hasScheduleDelayReject only fires at maxLocality == ANY with pending-but-unschedulable tasks. BIN_PACK just moves the repeated probing of a single offer from the outer loop into the inner loop; the final no-launch probe is what can set the reject in both cases, so noDelayScheduleRejects reaches the same result.
• Locality is preserved — BIN_PACK only applies at NO_PREF/ANY, and the PROCESS_LOCAL/NODE_LOCAL tests confirm it.

Two non-blocking notes:

1. Executor IDs are sorted lexicographically (sortBy(_.executorId) on the string), so e.g. "10" sorts before "2". That's fine for correctness — bin-packing only needs a stable order — and the doc does say "lexicographic executor ID order," so it's deliberate. Just flagging that since executor IDs are numeric-looking, the ordering can read as surprising; a half-sentence noting it's lexicographic purely for determinism would help the next reader.
2. Under BIN_PACK the inner loop already fills each offer to exhaustion, so the enclosing do/while(launchedTaskAtCurrentMaxLocality) will make one extra scan that launches nothing before exiting. Harmless, just a minor redundant pass — not worth changing.

Docs and test coverage are thorough. LGTM.

[pan3793]

this one looks very similar to #56957

[cloud-fan]

So the motivation seems to make auto-scale more efficient? Can we have a proper design discussion for the end to end auto-scale workflow? Why changing the scheduler is the best design choice? Just for example, one idea can be: when the auto scaler predicate an upcoming idle time window, it can pick some executors and mark them as decommissioned, then scheduler will not assign new tasks to these executors, and auto scaler can kill them after current tasks are done.

[sunchao]

@pan3793 Interesting! I didn’t realize there was already another PR solving the same problem. It probably makes sense to consolidate the efforts rather than have two similar implementations. WDYT, @starcatmeow?

@cloud-fan Yes, the main motivation is to make dynamic allocation more efficient. Bin-packing seemed like a relatively straightforward way to do that while building on Spark’s existing scheduling and scale-down mechanisms.

one idea can be: when the auto scaler predicate an upcoming idle time window, it can pick some executors and mark them as decommissioned, then scheduler will not assign new tasks to these executors, and auto scaler can kill them after current tasks are done.

That’s an interesting idea. One question I have is how reliably Spark can predict future executor demand or an upcoming idle window. If demand picks up again shortly afterward, prematurely decommissioning executors could cause unnecessary churn.

I also see bin-packing as complementary to that approach: it concentrates tasks on fewer executors, leaving the others idle and creating more opportunities for scale-down. The autoscaler can then decide whether to remove those executors using its existing idle timeout or, in the future, a more demand-aware policy.

Happy to have more design discussions and hear more ideas on making auto-scaling better!

[starcatmeow]

@viirya Thanks for the thorough review and for checking these edge cases. Both non-blocking notes make sense. Since we may consolidate with #56957, I’ll keep them in mind for whichever implementation becomes the base.

Thanks @pan3793 for pointing out #56957, and thanks everyone for the discussion.

I reviewed #56957, and I agree that the two PRs overlap substantially and should be consolidated rather than maintained as two independent implementations.

The main differences I see are:

• [SPARK-57878][CORE] Add task assignment strategy to support binpack/balance task scheduling #56957 introduces a general TaskAssignmentStrategy abstraction with roundrobin, binpack, balance, and none. Its bin-packing strategy sorts the initial WorkerOffer.cores snapshot in ascending order and keeps assigning to the current offer until it is exhausted. Since WorkerOffer.cores represents free cores, this approximates busy-first placement when executors have uniform capacity.
• [SPARK-43086][CORE] Add configurable bin-packing task placement #57484 has a narrower scope: it keeps the existing spread behavior and adds only an opt-in bin-packing mode. It explicitly places executors that already have running tasks, or have received assignments earlier in the same resourceOffers call, before idle executors. It then uses a stable executor ID order within each group.
• [SPARK-57878][CORE] Add task assignment strategy to support binpack/balance task scheduling #56957 applies the selected strategy independently at every locality level, while [SPARK-43086][CORE] Add configurable bin-packing task placement #57484 limits bin-packing to NO_PREF and ANY. I think the broader locality behavior in [SPARK-57878][CORE] Add task assignment strategy to support binpack/balance task scheduling #56957 is also reasonable because TaskSetManager still gates every launch by locality. For PROCESS_LOCAL, the eligible executor is normally already fixed; for NODE_LOCAL, the strategy may change which executor on the same host is filled first, but it does not reduce locality.

I do not have a strong preference about which PR should be the base. If the maintainers prefer the more general assignment-strategy abstraction, I am happy to collaborate on #56957, port over any useful busy-state handling and test
coverage from this PR, and close this PR. If a narrower change specifically targeting SPARK-43086 is preferred, we can use #57484 as the base and incorporate the relevant details from #56957. I am happy to follow the maintainers' guidance on the desired scope and locality behavior.

@cloud-fan Happy to have a broader design discussion! I don’t want to claim that changing the scheduler is necessarily the best or only approach.
My main question with autoscaler-driven draining is how to predict future executor demand. The current executor target seems like the most direct signal, but future demand is highly workload-dependent. Proactively decommissioning based on the target may require additional hysteresis, such as using its historical trend and only scaling down after it remains low for some time.
The goal of bin-packing here is to reuse the existing executor idle timeout as a smoothing mechanism: after a wave of tasks finishes, executors are removed only if they remain unnecessary beyond the timeout, reducing scale-down/scale-up churn. This has worked well for our workloads, but I’m happy to discuss better end-to-end designs for dynamic allocation.

[dongjoon-hyun]

Thanks, @starcatmeow.

Yes, when it comes to duplicate or similar PRs, Apache Spark goes by a first-come, first-served rule. Given the code content of both PRs, I'd recommend to proceed @ulysses-you 's PR first.

• [SPARK-57878][CORE] Add task assignment strategy to support binpack/balance task scheduling #56957

[starcatmeow]

Thanks for clarifying. I’ll close this PR in favor of #56957, happy to collaborate there. Thanks everyone for the reviews and discussion!