# DDIA Ch. 11: one hot key sends 85% of a join to a single reducer, and 8× the reducers buys ~1× the speedup ## Concept Run a reduce-side join as a tiny in-process MapReduce: every event is keyed by `user_id` and shuffled to reducer `md5(user_id) % R`, where the matching reducer joins it against that user's profile row. With a **uniform** key distribution the R reducers get near-equal load (max/mean **1.02**), and doubling R halves the straggler's work — near-linear speedup. Now introduce **one** dominant key: a `celebrity` who accounts for **85%** of the events. Every one of those records hashes to the *same* reducer, which now holds **86.9%** of the whole job while the other seven idle. Then try the obvious fix — **add reducers** — and watch it do nothing: at R=64 the hot reducer still holds **85.2%** (852,410 records), because a key is *atomic* under hashing and always lands on one reducer. The real fix is to **salt** the hot key: rewrite `celebrity` into `celebrity#0..celebrity#15` on the fact side and replicate its dim row across all 16 salts. The salts hash independently, the straggler collapses from **868,616 → 178,614** records (4.9× lighter) on the *same* 8 reducers, and its hot-key share drops from 86.9% to 17.9%. ## Provenance *Designing Data-Intensive Applications*, 2nd ed. (Kleppmann & Riccomini, 2025), Chapter 11 — *Batch Processing*, §"Shuffling Data" and §"Reduce-Side Joins and Grouping": > ...a hash of the key ... determine[s] which reducer ... receives ... a > particular key ... all the records with the same key end up at the same > reducer. The book warns this is exactly where skew bites: *"if there is a very large amount of data associated with a single key ... [it] can lead to significant skew ... one reducer that must process significantly more records than the others."* This exercise turns that sentence into numbers you predict — the straggler, the failed attempt to fix it with more reducers, and the salting technique the book prescribes. ## Prerequisites The surprise is that a key is an *indivisible* unit of work under hashing, so one hot key is a serial bottleneck no amount of parallelism relieves. These help; each line notes where it shows up. 1. **Hash partitioning to reducers** — the shuffle sends record `k` to reducer `md5(k) % R`, so *every* record sharing a key meets at one reducer (that is what makes the join work). It shows up as the `partition()` function and the balanced Step 1 histogram. 2. **Data skew / hot keys** — real key frequencies are wildly unequal; one `celebrity` here is 85% of the events. It shows up in Step 2 as one reducer holding 86.9% of the job. 3. **The slowest reducer sets the wall-clock** — reducers run in parallel, so the job finishes when the *straggler* does; wall-clock ≈ the max reducer load, not the mean. It shows up as "wall-clock vs R=8" tracking the max column, not the total. 4. **Salting / replicating a hot key** — split the hot key into S sub-keys on the fact side and copy the matching dim row to all S salts, so its load spreads and the join still matches. It shows up in Step 4 as `celebrity#0..#15` and the collapsed straggler. ### Where to learn the prerequisites - **Hash partitioning & reduce-side joins (#1, #2):** DDIA §"Shuffling Data" and §"Reduce-Side Joins and Grouping" — the shuffle groups records by key at a reducer, and "handling skew" is called out by name. - **Skew / stragglers (#3):** DDIA §"Handling skew" — the *skewed join* / *hot-key* problem and why one heavy reducer stretches the whole job. - **Salting a hot key (#4):** the same technique as the Ch. 7 serving-time fix (see the [hot-key-skew exercise](../ch07/hot-key-skew.html)); Pig's *skewed join* and Hive's *skew join* automate it. If only one is new, make it **#3** — that "a hash sends all copies of a key to one reducer, and the slowest reducer is the wall-clock" is the whole result: it is why the hot partition can't be split and why more reducers don't help. ## Environment macOS 26.5.2 (Darwin 25.5.0), arm64; Python 3.15.0a8, standard library only; no Docker; seeded + stable md5 hashing, so the per-reducer counts reproduce exactly. - Workload assignment fixed by `random.Random(42)` - Partitioner is `int(hashlib.md5(key.encode()).hexdigest(), 16) % R` — **not** Python's built-in `hash()` (which is salted per process and would not reproduce) - 1,000,000 events; the celebrity draws ~85% of them; the tail is 10,000 users - Wall-clock is modeled as the max reducer's record count (unit time per record, reducers in parallel), so "speedup" is `max_load(R=8) / max_load(R)` ## Mental model One shuffle, three workloads laid on top of it. - **The shuffle** — each event's key is hashed to `md5(key) % R`; that reducer receives every record for the key and joins it to the key's one dim row. A key is therefore an **atomic** unit: all of its records go to exactly one reducer, and hashing can never split them. - **Uniform keys** — 10,000 users, each roughly equally likely. Load divides evenly across the R reducers, so the max reducer ≈ N/R and doubling R halves it. Parallelism works as advertised. - **One hot key** — `celebrity` is 85% of the events. All 850k of its records hash to one reducer. Because the wall-clock is the *slowest* reducer, the job is now ~85% serial: seven reducers finish in a blink and idle while the eighth runs almost the whole job. Adding reducers shrinks the *tail's* share of each reducer but leaves the celebrity's 850k glued to one reducer — the straggler is stuck. - **Salting** — turn the one hot key into S keys (`celebrity#0..celebrity#15`) by appending a random suffix on the fact side, and replicate the celebrity's dim row to all S salts so the join still matches. Now the 850k records split across up to S reducers. You have broken the atomicity — at the cost of S copies of the dim row and a recombine step for anything that reads the celebrity back. The whole thing is one deterministic script, `code/skewed_join_hot_reducer.py`. ## Setup ``` cd study/ddia/ch11 python3 code/skewed_join_hot_reducer.py ``` Do **not** read the output yet — make each prediction first. ## Steps ### Step 1 — uniform keys: parallelism works **Predict.** 1,000,000 events over 10,000 uniformly-random users, partitioned to R=8 reducers by `md5(user_id) % 8`. How even is the per-reducer load (max/mean)? And if you go from R=8 to R=16 to R=64, how does the straggler's load — the wall-clock — move? **Observe.** ``` ==================================================================== STEP 1 -- UNIFORM keys: reduce-side join, partition by md5(key) % R ==================================================================== 1,000,000 events, 10,000 distinct users, no hot key. R = 8 reducers: reducer 0 | 124,292 ####################################### reducer 1 | 124,517 ####################################### reducer 2 | 127,760 ######################################## reducer 3 | 121,949 ###################################### reducer 4 | 124,496 ####################################### reducer 5 | 125,236 ####################################### reducer 6 | 127,849 ######################################## reducer 7 | 123,901 ####################################### max=127,849 mean=125,000 max/mean = 1.02 (balanced: the hash spreads load evenly) Same uniform workload, more reducers -- wall-clock = the straggler (max load): R | max reducer load | max/mean | wall-clock vs R=8 8 | 127,849 | 1.02 | 1.00x 16 | 66,736 | 1.07 | 1.92x 64 | 18,539 | 1.19 | 6.90x -> doubling R roughly halves the max load: near-linear speedup. ``` **max/mean = 1.02, and 8× the reducers gives ~6.9× the speedup.** With no hot key, load divides evenly and parallelism pays off almost linearly. This is the world the MapReduce speedup story assumes. ### Step 2 — one hot key floods one reducer **Predict.** *Same shuffle, same R=8.* Now 85% of the events belong to a single `celebrity` user. What fraction of the whole job lands on the one reducer that holds `celebrity`? What is max/mean? **Observe.** ``` ==================================================================== STEP 2 -- SKEWED keys: one celebrity owns ~85% of the events ==================================================================== 1,000,000 events, but `celebrity` appears in 849,833 of them (85.0%). R = 8 reducers: reducer 0 | 868,616 ######################################## reducer 1 | 18,757 # reducer 2 | 19,277 # reducer 3 | 18,220 # reducer 4 | 18,872 # reducer 5 | 18,651 # reducer 6 | 18,916 # reducer 7 | 18,691 # hottest is reducer 0 (holds `celebrity`): 868,616 records = 86.9% of the whole job max=868,616 mean=125,000 max/mean = 6.95 (one reducer holds the wall-clock hostage) the other 7 reducers finish early and idle while it runs the job alone. ``` **Reducer 0 holds 868,616 records — 86.9% of the job — while the other seven hold ~18,800 each.** max/mean jumped from 1.02 to **6.95**. The seven light reducers finish almost instantly and then sit idle; the job's wall-clock is now essentially the time for one reducer to grind through 85% of the data. Eight reducers, but the work is ~85% serial. ### Step 3 — add reducers: does the straggler shrink? **Predict.** The obvious move: throw more reducers at it. Go R=8 → 16 → 64 on the *skewed* workload. Does the hot reducer's absolute load fall (does the job get faster)? What happens to max/mean? **Observe.** ``` ==================================================================== STEP 3 -- add reducers to the SKEWED job: does the straggler shrink? ==================================================================== R | hot reducer load | hot share | max/mean | wall-clock vs R=8 8 | 868,616 | 86.9% | 6.95 | 1.00x 16 | 858,826 | 85.9% | 13.74 | 1.01x 64 | 852,410 | 85.2% | 54.55 | 1.02x -> the hot reducer's ABSOLUTE load barely moves (~849,833 celebrity records always hash to ONE reducer). max/mean rises only because the mean shrinks. 8x the reducers buys ~1x the speedup: no help at all. ``` **8× the reducers buys ~1× the speedup.** The hot reducer's load goes 868,616 → 858,826 → 852,410 — it barely moves, because those ~849,833 `celebrity` records all hash to the *same* reducer no matter how many reducers exist. Adding reducers only splits the tiny tail more finely, so max/mean *rises* (13.74, 54.55) purely because the mean collapses. The straggler is untouched. More parallelism cannot help a unit of work that refuses to divide. ### Step 4 — salt the hot key **Predict.** Rewrite the hot key on the fact side: `celebrity` becomes one of `celebrity#0..celebrity#15` (S=16 salts), and replicate the celebrity's dim row across all 16 salts so the join still matches. *Same R=8.* What does the straggler's load drop to, and what is the new max/mean? **Observe.** ``` ==================================================================== STEP 4 -- SALT the hot key: `celebrity` -> `celebrity#0..#15` (S = 16), replicate dim ==================================================================== Same 1,000,000 events, same R = 8 reducers, hot key split into 16 salts: reducer 0 | 124,822 ############################ reducer 1 | 124,621 ############################ reducer 2 | 72,304 ################ reducer 3 | 177,447 ######################################## reducer 4 | 178,614 ######################################## reducer 5 | 18,651 #### reducer 6 | 125,407 ############################ reducer 7 | 178,134 ######################################## max=178,614 mean=125,000 max/mean = 1.43 (was 6.95) straggler load 868,616 -> 178,614 (4.9x lighter) hot-key share of the busiest reducer 86.9% -> 17.9% the 16 salts landed on 7 of 8 reducers, so the celebrity's load now spreads instead of piling up. COST: the dim row for `celebrity` is replicated across all 16 salts (a read/recombine must re-union them). ``` **The straggler collapses from 868,616 to 178,614 — 4.9× lighter — on the same 8 reducers**, and max/mean falls from 6.95 to **1.43**. Splitting the one hot key into 16 independently-hashed sub-keys let its load spread across 7 of the 8 reducers. It's not perfectly flat (16 salts don't divide evenly into 8 distinct reducers, so some reducers caught 3 salts at ~178k while reducer 5 caught none and kept only its 18,651 tail records) — but the serial bottleneck is gone. The cost: the celebrity's dim row now exists in 16 copies, and reading the celebrity back means re-uniting all 16 salted groups. ## What you should see - **Uniform** (Step 1): max/mean = **1.02**; R 8→64 gives max load 127,849 → 18,539, a **~6.9×** speedup — parallelism works. - **Skewed** (Step 2): one reducer holds **868,616** records = **86.9%** of the job; max/mean = **6.95**; the celebrity is **849,833** of 1,000,000 events (85.0%). - **More reducers don't help** (Step 3): R 8→64 leaves the hot reducer at **852,410** (still 85.2%); wall-clock speedup **~1.02×** for 8× the reducers. - **Salting** (Step 4): straggler **868,616 → 178,614** (4.9× lighter) on the same 8 reducers; max/mean **6.95 → 1.43**; hot-key share **86.9% → 17.9%**; the dim row is replicated across all **16** salts. ## Why MapReduce's speedup story rests on one assumption: that the work divides evenly across reducers. The shuffle sends record `k` to reducer `md5(k) % R`, which is a *promise about grouping* — every record for a key meets at one reducer so the join can happen — and, incidentally, a near-uniform spread when keys are distinct and roughly equally frequent. That is Step 1: 10,000 users over 8 reducers land ~125,000 each, and because each reducer runs in parallel and the job finishes with the *slowest* one, halving each reducer's share by doubling R halves the wall-clock. 6.9× faster for 8× the reducers. The hot key breaks the assumption at its root. `md5("celebrity")` is a single fixed value, so *all* 849,833 celebrity records hash to *one* reducer — the key is an **atomic** unit of work, indivisible by hashing. That reducer now holds 86.9% of the data while the mean is 12.5%: max/mean 6.95. And since wall-clock is the max, not the mean, the job is ~85% serial. This is why Step 3's fix fails: adding reducers only repartitions the 15% tail into ever-smaller slivers; the celebrity's 850k stay welded to one reducer, so its absolute load (868,616 → 852,410) and thus the wall-clock barely move. max/mean *climbs* to 54.55 not because the straggler got worse but because the mean fell — a vivid reminder that max/mean measures imbalance, and the number that sets your bill is the max. **You cannot parallelize past an atomic unit of work.** Salting attacks the atomicity itself. Appending a random suffix turns the one key `celebrity` into S=16 keys `celebrity#0..celebrity#15`, and because each suffix hashes independently, the 850k records split into ~16 groups that scatter across reducers — the straggler drops to 178,614, a 4.9× win on the *same* 8 reducers. The price is paid on the dim side: the join only matches if the celebrity's profile row exists under every salt, so it is replicated S times, and anything that later reads "the celebrity" must re-union all 16 salted groups. You have traded a modest amount of replication and a recombine step for the ability to split an otherwise-atomic key. **The boundary — uniform keys need no salting, and salting only pays for the few known hot keys.** Step 1 is the boundary made concrete: with no hot key, hashing already balances load and adding reducers scales linearly, so salting would only manufacture dim-row replication and recombine cost for no benefit. Salting is a *targeted* patch — you apply it to the handful of keys you know are hot, and only for joins/aggregations that tolerate the replicate-and-recombine (a `null`/default key that swallows unmatched rows is the other classic offender). Everywhere else it is pure overhead. The general law: a hash gives you *grouping* for free and *balance* only when the key frequencies are already even; a single skewed key is a serial bottleneck that more machines cannot touch, and the only way out is to stop treating it as one key. ### Go deeper 1. **Sweep the salt count.** Try S = 8, 32, 64 and watch max/mean approach 1.0 — then note the diminishing return past S ≈ R and the growing dim-replication cost. Where is the knee? 2. **Move the effect to serving time.** The identical phenomenon at *read* time — one celebrity overwhelming one shard, and the same salting fix — is the Ch. 7 [hot-key-skew exercise](../ch07/hot-key-skew.html). Compare the trade: batch salting pays in dim replication; serving-time salting pays in read fan-out. 3. **Sidestep the shuffle entirely.** For a *small* dim side, a map-side / broadcast join ships the whole dim table to every mapper and joins locally — no shuffle, so no reducer for the hot key to overwhelm. It is the sibling reduce-side-vs-broadcast join exercise (DDIA §"Map-Side Joins"): the hot-key straggler is a reduce-side problem that broadcast simply doesn't have. ## What this shares with the Ch. 7 hot-key exercise (and how it differs) Both are the *same phenomenon* — one hot key that hashing sends to one place, fixed by the same salting trick — but at opposite ends of the pipeline. Ch. 7's [hot-key-skew](../ch07/hot-key-skew.html) is a **serving-time** hot shard: a celebrity's *requests* pile onto one shard, and splitting the key spreads writes but amplifies reads. This exercise is the **batch-job** twin: a celebrity's *records* pile onto one reducer, the job's wall-clock waits on that straggler, and the same split spreads the load at the cost of dim-row replication. Read them together to see that "a hash makes a key atomic, so one hot key is indivisible" is a single idea that resurfaces wherever a hash decides placement.