# DDIA Ch. 11: a broadcast join moves 124x less data than a reduce-side join — for the same answer

## Concept

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**.

## Provenance

*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.

## Prerequisites

The magnitude is set by *which table crosses the network*. These help; each line
notes where it shows up.

1. **MapReduce = map → shuffle → reduce** — a map function emits key-value pairs;
   the framework *shuffles* (partitions + sorts) so all pairs with the same key
   land at one reducer; the reduce function processes each key's group. It shows
   up as the mapper/partitioner/reducer pipeline both joins are built from.
2. **Partitioned / sort-merge (reduce-side) join** — to join two tables, emit
   every row of both keyed by the join key; the shuffle co-locates matching rows
   at a reducer, which merges them. It shows up as `reduce_side_join`, whose
   shuffle carries *both* whole tables.
3. **Broadcast / map-side (hash) join** — replicate one table to every mapper,
   load it into an in-memory hash table, and probe it while scanning the other
   table locally; no shuffle of the big table. It shows up as `broadcast_join`,
   whose shuffle carries only `N_mappers` copies of the dim.
4. **"Small enough to broadcast"** — the broadcast side must fit in each mapper's
   memory. 62 KB does; 60 MB might; a second 60 MB fact table would not — that
   is the boundary where you fall back to a reduce-side join.

### Where to learn the prerequisites

- **MapReduce and the shuffle (#1–#2):** DDIA §"MapReduce Workflows" and
  §"Reduce-Side Joins and Grouping"; Dean & Ghemawat, "MapReduce: Simplified
  Data Processing on Large Clusters" (2004) for the map/shuffle/reduce model.
- **Map-side / broadcast hash joins (#3–#4):** DDIA §"Map-Side Joins" —
  broadcast hash joins, partitioned hash joins, and map-side merge joins, and
  the memory condition that lets a table be broadcast.

If only one idea is new, make it the **shuffle-cost model** (#2 vs #3): the byte
count is *whichever table you move*, and that single choice carries the entire
124× result.

## Environment

Seeded, so the byte counts reproduce exactly:

- macOS 26.5.2 (Darwin 25.5.0), arm64
- Python 3.15.0a8, standard library only — no Docker, no deps
- No real cluster: mappers, the partitioner, the shuffle byte counter, and
  reducers are all simulated in one process, so the counts are fixed every run
- Fixed workload: 1,000,000 fact events over 1,000 user ids, joined to 1,000
  dim profiles; `random.Random(42)`

## Mental model

Two joins, same two tables, same 1,000,000-row answer — only *what crosses the
network* differs.

- **Reduce-side (sort-merge) join** — a join needs matching rows in the same
  place. This one *moves both tables to a common reducer*: map emits every fact
  row and every dim row keyed by `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.
- **Broadcast (map-side hash) join** — instead of moving the big side, it
  *replicates the small side to where the big side already lives*: the 62 KB dim
  is copied to each of `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.

## Setup

```
cd study/ddia/ch11/code
python3 shuffle_vs_broadcast_join.py
```

Do **not** read the output yet — make each prediction first.

## Steps

### Step 1 — the two tables

**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?

**Observe.**

```
Join a FACT table to a DIMENSION table with an in-process MapReduce.
    fact: 1,000,000 activity events  = 60.5 MB on disk
    dim:  1,000 user profiles    = 62.6 KB on disk
    the fact table is 990x larger than the dim table
```

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.

### Step 2 — the reduce-side (sort-merge) join

**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?

**Observe.**

```
reduce-side (sort-merge) join -- shuffle BOTH tables to reducers by user_id:
    rows joined      = 1,000,000
    bytes shuffled   = 60.5 MB  (= size(fact) + size(dim))
```

**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.

### Step 3 — the broadcast (map-side hash) join

**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?

**Observe.**

```
broadcast (map-side hash) join -- ship the dim to all 8 mappers, fact stays put:
    rows joined      = 1,000,000
    bytes shuffled   = 500.5 KB  (= 8 mappers x size(dim))
```

**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.

### Step 4 — same answer, 124× less data

**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?

**Observe.**

```
same join result?
    reduce-side: 1,000,000 rows, checksum 175720604590266420
    broadcast:   1,000,000 rows, checksum 175720604590266420
    IDENTICAL -> True

data moved over the network:
    reduce-side  60.5 MB
    broadcast    500.5 KB
    broadcast moves 124x LESS data for the SAME answer.
```

**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.

## What you should see

- The fact table is **60.5 MB**, the dim table **62.6 KB** — ~**990×** larger.
- The reduce-side join shuffles **60.5 MB** (`size(fact) + size(dim)`) and joins
  **1,000,000** rows.
- The broadcast join shuffles **500.5 KB** (`8 mappers × size(dim)`) and joins
  the same **1,000,000** rows.
- Both emit the **identical** checksum `175720604590266420` — broadcast moves
  **124× less data** for the same answer.

## Why

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.**

<div class="callout key">

**The boundary — broadcast wins only while the small side fits in memory.** The
formula `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.

</div>

## Go deeper

1. **Partitioned (bucketed) hash join.** If *both* tables are large but already
   partitioned the *same way* on the join key (same number of buckets, same hash),
   a reducer only needs the matching bucket from each side — you shuffle within a
   bucket, not the whole table. Pre-bucketing both inputs (Hive/Spark "bucketed
   tables") turns an unavoidable full shuffle into many small independent joins.
2. **Break the broadcast.** Grow the dim table (or `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.
3. **Skew.** This workload spreads events evenly over 1,000 users. Concentrate
   most events on one "celebrity" `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.