Join a 1,000,000-row fact table to a 1,000-row dimension table two ways. A reduce-side (sort-merge) join shuffles both tables across the network — 60.5 MB. A broadcast (map-side hash) join ships only the small dim to every mapper and joins the fact table in place — 500.5 KB. Same 1,000,000 joined rows, identical checksum, but 124× less data on the wire.
Join a 1,000,000-row fact table (activity events keyed by user_id) to a 1,000-row dimension table (user profiles) with a tiny in-process MapReduce, two ways. A reduce-side (sort-merge) join emits every record of both tables keyed by user_id and shuffles them across the network to reducers — so the bytes on the wire are size(fact) + size(dim): the whole big table moves. A broadcast (map-side hash) join ships the small dim table to every mapper once, builds a local hash, and joins the fact table in place — so the fact table never crosses the shuffle and the bytes on the wire are N_mappers × size(dim). Same 1,000,000 joined rows, identical checksum — but the broadcast join moves 124× less data.
Designing Data-Intensive Applications, 2nd ed. (Kleppmann & Riccomini, 2025), Chapter 11 — Batch Processing, §"Joins and Grouping" / §"Shuffling Data":
When a mapper emits a key-value pair, the key acts like the destination address to which the value should be delivered [...]. Bringing all the records with the same key together in the same place is the whole purpose of the sort in MapReduce.
The book contrasts reduce-side joins (sort and shuffle both inputs to a common reducer) with map-side joins (a broadcast hash join keeps a small enough input in memory on every mapper, avoiding the sort and shuffle of the large input entirely). Here you watch the byte counter separate the two by two orders of magnitude.
The magnitude is set by which table crosses the network. These help; each line notes where it shows up.
reduce_side_join, whose shuffle carries both whole tables.broadcast_join, whose shuffle carries only N_mappers copies of the dim.random.Random(42).
Two joins, same two tables, same 1,000,000-row answer — only what crosses the network differs.
user_id, the shuffle partitions them by key (fact row for user 42 and profile 42 land at the same reducer), and the reducer joins each group. Network cost = size(fact) + size(dim) — you pay to ship the entire big table.N_mappers mappers, loaded into a hash table, and each mapper probes it while scanning its local slice of the fact table. The fact table never crosses the shuffle. Network cost = N_mappers × size(dim).Both are one script, code/shuffle_vs_broadcast_join.py, with a real byte counter summing the serialized size of every record that crosses the shuffle boundary. Both must emit the identical joined output.
cd study/ddia/ch11/code
python3 shuffle_vs_broadcast_join.pyDo not read the output yet — make each prediction first.
Predict. The fact table is 1,000,000 events (user_id, event, item, ts); the dim table is 1,000 profiles (user_id, name, country, plan). Roughly how many times larger is the fact table than the dim table on disk?
60.5 MB vs 62.6 KB — the fact table dwarfs the dim table by ~990×. Hold that ratio: it is the ceiling on how much a broadcast join can save.
Predict. Map both tables to (user_id, tagged record) and shuffle to 16 reducers, which group by user_id and join. How many bytes cross the shuffle boundary — and how does that compare to the two tables' sizes?
60.5 MB — essentially the whole fact table. Every one of the 1,000,000 fact rows is emitted keyed by user_id and physically moved to a reducer, plus the 1,000 dim rows. The join is correct, but you paid to ship the entire big table across the network.
Predict. Now ship the 62.6 KB dim table to every one of 8 mappers, build a local hash, and join the fact table in place. How many bytes cross the shuffle now — and is the answer the same?
500.5 KB — just 8 copies of the tiny dim table. The fact table never moves. Same 1,000,000 rows joined, but the network carried three orders of magnitude less than the fact table's 60.5 MB.
Predict. Both joins claim 1,000,000 rows. If we checksum the joined output (order-independent), will the two strategies match — and what is the ratio of bytes shuffled, reduce-side over broadcast?
Identical checksum, 124× less data. A join is not a join: both produce the exact same output rows, but one moves 60.5 MB and the other 500 KB. The gap is whichever table you chose to ship.
size(fact) + size(dim)) and joins 1,000,000 rows.8 mappers × size(dim)) and joins the same 1,000,000 rows.175720604590266420 — broadcast moves 124× less data for the same answer.A join needs matching rows co-located: to combine fact row 42 with profile 42, they must be on the same machine at the same time. There are only two ways to arrange that, and they differ only in which data you move.
The reduce-side join moves both sides to a common partition. The map phase tags every record with its user_id and the shuffle uses that key as a destination address — so fact rows and the profile for user 42 all get routed to the same reducer, where the merge happens. The cost is the sum of everything that crossed the shuffle: size(fact) + size(dim) ≈ 60.5 MB + 62.6 KB ≈ 60.5 MB. The dim table is a rounding error; you paid, overwhelmingly, to relocate the entire fact table across the network — and you also paid to sort it, which this counter doesn't even bill.
The broadcast join inverts the move: instead of shipping the big side to a rendezvous point, it replicates the small side to wherever the big side already sits. Each of the 8 mappers receives one copy of the 62.6 KB dim, loads it into an in-memory hash table keyed by user_id, then streams its local slice of the fact table through, probing the hash for each event. The fact table never enters the shuffle at all. The cost is N_mappers × size(dim) = 8 × 62.6 KB ≈ 500.5 KB. The ratio is therefore roughly size(fact) / (N_mappers × size(dim)) = 60.5 MB / 500.5 KB ≈ 124× — and because both joins compute the same inner join on user_id, the output rows (and their order-independent checksum) are bit-for-bit identical. The intuition shortcut: the byte bill is whichever table you move, so move the small one.
N_mappers × size(dim) beats size(fact) + size(dim) exactly when N_mappers × size(dim) < size(fact), i.e. when the dim table is small enough (relative to the fact table and the mapper count) to replicate everywhere. That "small enough" is a hard memory limit: each mapper must hold the whole broadcast side in RAM. Make both tables large — join two 60 MB fact tables — and neither fits to broadcast, the map-side trick evaporates, and you are back to a reduce-side join shuffling both. Broadcast is not a faster join; it is a bet that one side is small, and the bet pays exactly the lopsidedness of the sizes.
N_mappers) until N_mappers × size(dim) crosses size(fact) and watch the ratio fall below 1× — the point where broadcasting is more expensive than shuffling, so you'd switch strategies.user_id and a reduce-side join sends them all to one overloaded reducer (a hot key) — the straggler problem the next exercise builds, and a case where even a "cheap" join stops finishing on time.Sources: DDIA 2e, Ch. 11, §"Joins and Grouping", §"Reduce-Side Joins and Grouping", §"Map-Side Joins" · Dean & Ghemawat, "MapReduce: Simplified Data Processing on Large Clusters" (2004).