Design a shuffle
A distributed job must repartition data: every producer task sends a slice of its output to every consumer task, for a join, a group-by or a sort. With M producers and R consumers that is M × R connections, and as the job grows each one carries fewer bytes. How wide should the shuffle be, and how should it be written and moved?
You will end with a shuffle width, a writer (hash or sort), a placement, and compression and serialization settings, each justified by arithmetic.
Work through it
Compute the bytes per connection
Bytes per connection = total shuffle bytes ÷ (producers × consumers). Compare it with the smallest transfer the storage reads efficiently, roughly its latency × bandwidth:
| Medium | Efficient transfer |
|---|---|
| Memory or RDMA | about 10 KB |
| Local NVMe | about 250 KB |
| Spinning disk | about 2 MB |
| Object storage | about 4 MB, plus per-request charges |
If you are below it: you are limited by operations per second, and more bandwidth does not help. 1 TB over 10,000 × 10,000 tasks is 10 KB per connection: 200 times below a disk's floor.
Cap the width
Setting bytes per connection at the floor gives the widest useful square shuffle: producers = consumers ≤ √(total bytes ÷ floor). 1 TB over a 1 MB floor gives 1,000 × 1,000. If your partition count is above that: you pay for parallelism you cannot use. Shuffle width grows only with the square root of the data.
Shrink the data before changing the topology
Pre-aggregate on the producer side when each task has many rows per distinct key. Confirm that column pruning and filters really happened upstream. Broadcast the small side instead of shuffling when consumers × small side < large side. Ten times less data also raises the width cap by √10.
Choose the writer by consumer count
Low hundreds of consumers: hash shuffle, writing straight to one file per consumer. More: sort shuffle, one sorted file per producer with an index. The crossover is set by open files and buffer memory: cores × consumers × buffer size.
If the consumer needs order, sort on the producer side
Sorting by key while writing makes the order free for the consumer; stop counting it as a shuffle cost.
Decide whether to use a shuffle service
Use a push-based or external shuffle service when bytes per connection are far below the floor, when executors must be released (spot instances, autoscaling, scale to zero), when losing a node must not recompute finished stages, or when you need to split skewed partitions.
Use a tree only for reductions
A tree of combining steps helps associative reductions and always helps the final gather to one node. For a repartitioning shuffle it buys nothing.
Compress only if it pays at the real link speed
Compress if and only if compression speed > link speed × ratio ÷ (ratio − 1), at the actual link speed, not the one in the tuning guide.
Never use a general-purpose serializer
Ship the engine's internal representation; converting to and from a generic format costs more than it saves.
Make the partition function deterministic
Before enabling retries or speculation, verify that the same record always goes to the same consumer. Otherwise a retried task sends data to different places than the first attempt did.