The shuffle: a distributed transpose
Why does moving rows between stages cost a product of two widths, and what governs that cost?
Why the shuffle exists
The first three articles asked what a boundary costs, and how work is divided. The shuffle exists because those two answers conflict:
A shuffle is not a data-movement mechanism that happens to be expensive. It is the price of changing your mind about the partitioning key. It is also the only operation here whose cost is a product of two stage widths, not a function of data size alone.
The shape of the problem is a transpose. Think of a table with M rows and R columns. Each of M producers holds R fragments, one for each consumer. Each of R consumers needs M fragments, one from each producer. You must turn a layout that is contiguous by producer into one that is contiguous by consumer. (M x R is the notation used from now on.)
On storage with a minimum efficient transfer size, a transpose is a granularity problem by construction. The natural write is large and sequential (each producer writes one big output). The natural read is M small pieces for each consumer. Every mechanism in this article is an argument about who absorbs that mismatch.
The two numbers that govern everything here
Number 1: bytes per connection
Compare b with the substrate's minimum efficient transfer: the size at which you get most of the sequential bandwidth despite the access latency. It is roughly latency × bandwidth.
| Substrate | Access latency | Per-stream bandwidth | Efficient transfer (about 50% of peak) |
|---|---|---|---|
| Memory / RDMA (remote direct memory access over a fast network) | ~1 µs | ~10 GB/s | ~10 KB |
| Local NVMe | ~80 µs | ~3 GB/s | ~250 KB |
| Spinning disk | ~10 ms | ~175 MB/s | ~2 MB (about 16 MB for 90%) |
| Object storage | ~50 ms | ~80 MB/s per connection | ~4 MB, plus ~$5 per million write requests |
Check the arithmetic. NVMe: 80 µs × 3 GB/s = 240 KB, about 250 KB. Disk: 10 ms × 175 MB/s = 1.75 MB, about 2 MB. Object storage: 50 ms × 80 MB/s = 4 MB.
Mateo's numbers. Suppose the shuffle moves S = 1 TB over local disks (floor about 2 MB). With M = R = 10,000: b = 10¹² / (10⁴ × 10⁴) = 10,000 bytes = 10 KB per connection. That is 200 times below the 2 MB floor. The disks do small random reads and spend their time on positioning, not on moving data.
Number 2: the width ceiling
Set b equal to the substrate's floor b_min and solve for the width. M × R ≤ S / b_min. For a square shuffle (M = R):
Shuffle width grows as the square root of the data size.
- 1 TB over a 1 MB floor: √(10¹² / 10⁶) = √10⁶ = 1,000, so 1,000 × 1,000.
- 100 GB: √(10¹¹ / 10⁶) = √10⁵ ≈ 316.
- 10 TB: √(10¹³ / 10⁶) = √10⁷ ≈ 3,162.
Ten times the data buys only about three times the parallelism. This is not a tuning artifact. It is the geometry of a transpose over a granular substrate.
Every mechanism below tries to do one of two things, because only two variables in the formula can be changed by engineering:
- lower b_min: merge fragments before anyone reads them; or
- shrink S: reduce the data before moving it.
The exchange operator
The shuffle appears in the plan as a single node, the exchange. That design decision is what makes parallel execution manageable. An exchange holds a partitioning function. Producers call it for each row and write the row into a per-target buffer. Consumers read their own partition. Every other operator stays unaware of parallelism.
Three jobs at one node. This is the real reason the shuffle gets first-class status:
- It is the parallelism mechanism (the only place where partitioning is decided).
- It is a pipeline breaker (so it defines the stage boundary).
- If its output is kept, it is the fault-tolerance substrate (the only point in the plan from which data can be read again after a failure).
No other part of the runtime carries three unrelated responsibilities. Most shuffle design tension comes from this: improving one of the three moves the other two.
The partitioning function has preconditions. It is a design choice, not a detail:
| Function | What it does | Notes |
|---|---|---|
| Hash | Balances by key count, not by size | Correct by default; balanced only if keys are uniform. The default everywhere, and the source of all skew. |
| Range | Required when the consumer needs global order | Boundaries depend on the data distribution, so it needs a sampling pass first (reservoir-sampled boundaries are the standard answer; a reservoir sample is a fixed-size random sample taken in one pass). |
| Round-robin | Perfect balance, no co-location | Legal only when the consumer does not care about keys (a repartition before a write, a rebalance before a map-only stage). Illegal before a join or a group-by. |
| Broadcast / replicate | No partitioning at all | A topology of its own; see the end of this article. |
| Gather (R = 1) | The final collect | The maximal incast case (many senders to one receiver). |