andygrove opened a new issue, #2319: URL: https://github.com/apache/datafusion-ballista/issues/2319
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** Consumer tasks are assigned to executors without regard for where their input already lives. The scheduler's built-in distribution policies are `Bias` (fill each executor's slots eagerly) and `RoundRobin` (spread evenly), neither of which looks at the `PartitionLocation`s the task will read. A reduce task therefore frequently lands on an executor that holds none of its input and fetches every block over Arrow Flight, even when one executor already holds most of it on local disk. `ShuffleReaderExec` already distinguishes the two paths, reading directly from the filesystem when a partition is local and going over Flight otherwise (unless `ballista.shuffle.force_remote_read` is set), so the win from placing the task well is immediate: local reads instead of network round trips, less pressure on the shuffle governor, and fewer connections to the producing executors. **Describe the solution you'd like** An affinity-aware task distribution policy: when binding a task that reads shuffle input, prefer an executor that already holds the largest share of that task's input bytes, subject to available slots, and fall back to the current behavior when no candidate has a meaningful share or the preferred executor is saturated. This does not need a core change to prototype. Task distribution is already pluggable via `TaskDistributionPolicy::Custom(Arc<dyn DistributionPolicy>)` (#1243), so an affinity policy can be implemented and measured against `Bias` and `RoundRobin` before deciding whether it belongs in the scheduler as a built-in option. Open questions worth answering in a prototype: - How much locality is actually available in practice? With N producer tasks spread across the cluster, a reduce partition's input is usually spread too, so the best candidate may only hold 1/N of the bytes. Affinity likely matters most at low executor counts, after AQE coalescing, and for pass-through (non-repartitioning) stage boundaries. - What is the cost of the resulting slot idling? Waiting for the preferred executor's slot trades network for latency, so the policy needs a bound on how long it will hold out. - Interaction with straggler behavior and with `ballista.shuffle.reader.max_bytes_in_flight`. **Describe alternatives you've considered** - Leave placement alone and make remote reads cheaper (`IO_BLOCK_TRANSPORT`, client pooling, larger blocks). Already partly done and complementary, but it cannot beat not making the request at all. - Reduce the number of blocks per task instead, via a cheaper shuffle format (#660). Also complementary. **Additional context** - Discussed on #2308; suggested by @milenkovicm, who noted the pluggable policy hook makes this cheap to prototype. - Design rationale for the current model: `docs/source/contributors-guide/shuffle.md`. -- 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]
