starcatmeow opened a new pull request, #57484:
URL: https://github.com/apache/spark/pull/57484
### 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](https://issues.apache.org/jira/browse/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:
```properties
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:
```bash
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:
```bash
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`:
<img width="1280" height="720" alt="SPREAD Event Timeline"
src="https://github.com/user-attachments/assets/7dc908e4-619d-43a5-9933-4ce7628c111c";
/>
`BIN_PACK`:
<img width="1280" height="720" alt="BIN_PACK Event Timeline"
src="https://github.com/user-attachments/assets/b341ffbd-6cdb-4bb8-8350-617b26a53475";
/>
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
