In a distributed plan the operators are the easy part; the exchanges between them are where bytes are serialized, sent, spilled and deserialized, where one oversized partition becomes one straggler task, and where a wrong size estimate becomes an out-of-memory error. The order of moves is fixed: first move fewer bytes — prune partitions and columns, aggregate partially before the exchange (the MapReduce combiner), ship the small side instead of the large one — and only then think about moving them faster.
What the evidence says the cost is. MapReduce was designed around scarce bandwidth: tasks are scheduled next to their input, and the combiner merges map output before it crosses the network. A decade later, blocked-time analysis of Spark on two SQL benchmarks and a production workload found the opposite for the wire itself: a faster network improves job completion time by a median of at most 2%, CPU is often the bottleneck, for some queries up to half of CPU time goes to deserialization and decompression, and the median shuffled-to-input ratio was below 0.35. Both findings point the same way: the price of movement is mostly CPU to encode and decode, disk to spill, and skew — so fewer bytes and fewer exchanges help, a faster network mostly does not. Measure before choosing which (37, 89).
| Join strategy | What moves | Breaks when |
|---|---|---|
| Broadcast hash | the small side, to every task; the large side stays put | the size estimate is wrong and the “small” side does not fit in memory (Spark auto-broadcasts below autoBroadcastJoinThreshold, 10 MB by default) |
| Shuffle hash | both sides, hash-partitioned on the key | a partition’s build side does not fit its in-memory hash table — skew makes that likely, which is why Spark’s AQE switches to it only when every post-shuffle partition is below a threshold |
| Sort-merge | both sides, partitioned and sorted | it pays for sorting both sides, and a skewed partition still makes one straggler — the case Spark’s adaptive skew-join splitting targets |
| Storage-partitioned / bucketed | nothing — both sides are already laid out by the join key, so the plan has no exchange before the join | the layouts were not decided together at write time (92) |
Adaptive query execution (on by default in Spark since 3.2) re-plans at each shuffle boundary from the sizes it actually observed: it coalesces small partitions, turns a sort-merge join into a broadcast join, and in a sort-merge join splits a partition treated as skewed — larger than 5× the median and than 256 MB by default. It repairs estimates, not the data model: a hot key is still one key (92). In a streaming engine the exchange never ends — a regular Flink SQL join keeps both inputs in state for good unless a TTL bounds it, which is a state decision (82), not a tuning knob.
-- orders is PARTITIONED BY (order_date); dim_country is a few hundred rows
EXPLAIN FORMATTED
SELECT /*+ BROADCAST(c) */ c.region, sum(o.amount_cents) AS revenue_cents
FROM orders o
JOIN dim_country c ON o.country_code = c.code
WHERE o.order_date >= DATE '2026-09-01'
GROUP BY c.region;
-- Read the plan for movement, not for operators:
-- Scan orders PartitionFilters: (order_date >= 2026-09-01) older days never read
-- BroadcastExchange dim_country the small side travels
-- BroadcastHashJoin orders is not shuffled
-- HashAggregate partial_sum(amount_cents) combine before the exchange
-- Exchange hashpartitioning(region, 200) the only shuffle left
-- HashAggregate sum(amount_cents)
-- Every Exchange node is where rows are serialized and moved between tasks;
-- a date filter that does not show up under PartitionFilters reads every partition.