Remote shuffle, tree exchange and broadcast

Which designs reduce the damage on the read side: a push-based service, a tree exchange, or a broadcast?

Push-based and remote shuffle services

Everything above improves the write side. The read side is untouched: even with a perfect sort shuffle, each consumer issues M fetches of S/(M·R) bytes each. With M = 10,000 that is ten thousand small reads for each consumer. It is the irreducible half of a transpose: the producer can consolidate across its own targets, but it cannot consolidate across its peers.

The fix is to let a third party do the peer-side merge before the consumer arrives. Producers push per-partition blocks to a service (or to elected "merger" nodes), which appends them into one file per partition. The consumer then does one large sequential read.

Fetch count per consumer falls from M to about 1.
Read size rises from S/(M·R) to S/R, an improvement of M times in read granularity: four orders of magnitude in a wide shuffle (M = 10,000).

This is the largest single improvement anyone has made to the shuffle in the last decade. Several organizations built it independently at about the same time, which is the usual sign that a constraint, not a preference, was being addressed.

The costs, in order of how often they are underestimated

The merge work moved to the service, which now needs throughput proportional to your total shuffle rate. The relocation lands on a component that needs its own capacity planning.
There is an extra hop. Bytes now cross the network twice (producer to service, service to consumer) instead of once. A design that improves read granularity by M× costs 2× the network bytes. That trade is clearly right when b is 10 KB and clearly wrong when b is already 4 MB.
The service is stateful and shared, so its failure is correlated across every query that uses it. Merged data must either be replicated or keep a fallback path to the original unmerged blocks. Keeping both (merged for speed, unmerged for recovery) is the standard answer. It pays storage for availability, and the fallback is "fail-degraded" (slower but working) rather than "fail-fast".

Why remote shuffle really won: disaggregation

The reason remote shuffle won is not read granularity. It is disaggregation (shuffle data no longer lives on the compute node), and the causal order should be stated correctly. If shuffle data is not on the compute node:

an executor can be released as soon as its tasks finish;
losing a node does not force the recomputation of a completed stage;
disk capacity is sized independently of CPU.

That makes dynamic allocation, scale-to-zero and spot instances (cheap machines that the cloud may take back at short notice) workable for shuffle-heavy jobs. That is worth more money than the read-granularity win. Remote shuffle is mainly an elasticity and cost mechanism that also happens to fix the read pattern.

Object storage as the shuffle medium

Taking disaggregation all the way to object storage runs straight into the two numbers: the efficient-transfer floor is about 4 MB (tens of MB for 90% efficiency), and each write request costs about $5 per million. Request cost per gigabyte = $5 / (10⁶ × object size in GB).

100 KB objects: 1 GB = 10,000 objects, 10,000 × $5 / 10⁶ = $0.05 per GB (five cents).
10 KB objects: 100,000 objects per GB, $0.50 per GB (fifty cents).

That is why the successful designs are services with local disk and memory in front of object storage, not raw object-store shuffle.

Pull versus push, again

Control flow = flow control has its clearest instance here. Classic shuffle is pull: the consumer fetches when it is ready, so its own memory paces the transfer and backpressure is free. Push-based shuffle is push: the producer sends, so the service must absorb the rate, and an explicit flow-control scheme has to appear. Same trade, same direction, same price, once more. Pull also has a size to set: by Little's Law, in-flight fetch bytes must equal bandwidth × round-trip time to fill the link, so at 10 Gbps (1.25 GB/s) and 1 ms RTT that is 1.25 MB, and across an availability zone (about 5 ms RTT) about 6 MB. The in-flight fetch limit in every engine is really sizing this number.

A rule with predictive power. When a fan-out product exceeds the substrate's efficient granularity, an intermediate tier appears whose only job is to re-aggregate fragments. Examples: shuffle services; tree exchange (next section); the runs of an external merge; multi-level aggregation; hierarchical all-reduce in machine-learning training. It predicts that any new disaggregated architecture will grow a merge tier whether its designers planned one or not. That is what happened to object-store shuffle.

All-to-all, tree-structured exchange, and broadcast

Three topologies, distinguished by what each requires of the operation.

All-to-all

All-to-all requires nothing and pays the product: M·R connections, each carrying S/(M·R) bytes, with a fan-in of M at every consumer. That fan-in is incast (many senders answering one receiver at the same moment, which overflows switch buffers and collapses throughput). Here it is designed in rather than accidental. In terms of the Universal Scalability Law (a model in which coordination cost grows as N² and eventually reduces throughput), the shuffle is the coherency term made structural: the coordination cost is the topology.

Tree-structured exchange

Tree exchange puts levels of intermediate combination between producers and consumers. With fan-in f, the depth is log_f M, there are M connections per level, and the consumer's fan-in falls from M to f.

Its precondition is algebraic. Intermediate nodes can combine only if the operation is associative. Otherwise they can only forward, and you have added log_f M round trips of latency for no reduction. So tree exchange is legal for reductions and useless for repartitioning: a join's shuffle cannot be tree-reduced, because nothing can be combined on the way.

A precision, because the usual claim is loose. A tree does not move fewer bytes than all-to-all in general. If each level reduces the data by factor r, total bytes are S(1 + 1/r + 1/r² + …) ≈ S · r/(r − 1), and for large r that is about S. Example: r = 10 gives S × 10/9 ≈ 1.11 S. The win is in connection count, fan-in and the size of the final gather, not in volume (except where reduction shrinks S, which is the combiner's contribution, not the topology's). The tree wins clearly at R = 1, where all-to-all degenerates into maximal incast.

Broadcast

Broadcast is the only topology that avoids the transpose completely: send the whole input to every consumer, S · R bytes in R large sequential transfers. It trades total volume for granularity, and it removes the shuffle on one side of a join. So the comparison is between broadcasting the small side and shuffling both sides.

Broadcast cost: R · S_small bytes.
Shuffle both sides: S_small + S_large bytes.
Broadcast wins roughly when R · S_small < S_small + S_large, that is, when (R − 1) · S_small < S_large.

This is the familiar "broadcast join threshold", derived instead of quoted. It is expressed in absolute bytes in every engine's configuration because R and S_large are implicitly held constant.

Broadcast has its own topology question. R separate sends from one source make the source's fan-out the bottleneck, so engines spread a broadcast through a chain or a tree. The tree argument applies again, and here with no need for associativity, because forwarding is always legal.

Case, resolvedConstructed case · Mateo at Gridlark

Too many open files

A 10,000-by-10,000 shuffle on Lark first died with "too many open files", then crawled through millions of tiny reads. Where it started: The shuffle →

Mateo computes b = 1 TB / (10,000 × 10,000) = 10 KB, 200 times below the 2 MB floor of his disks. The width ceiling is √(10¹² / 2 × 10⁶) = √(500,000) ≈ 707. His 10,000-wide shuffle is more than 14 times wider than the substrate can use. The "too many open files" failure was the hash shuffle's R descriptors per task. The "millions of tiny reads" was the consumer side: every consumer issued 10,000 tiny fetches.

His steps follow the order of attack:

Attack S first. Partial aggregation on meter_id shrinks the output (each task sees about 1,000 rows per distinct key, a ratio far above 1), and projection pushdown drops unused columns. A 10× smaller S raises the width ceiling by √10 ≈ 3.2.
Broadcast the small table. tariffs (about 1 MB) is broadcast to all consumers, so the join with tariffs needs no shuffle at all (10 GB moved instead of 1 TB shuffled).
Lower the width toward the ceiling (a few thousand partitions rather than 10,000), and switch from hash to sort shuffle (one file and one index per task, one buffer, one descriptor).
Use an external shuffle service to merge across producers, so each consumer reads one large range instead of 10,000 small ones. As a bonus, executors can be released early and spot machines are safe to lose.

The job runs, and the disks now do few, large reads.