# DDIA Ch. 7: a hash spreads keys evenly across 16 shards, but one hot key still owns 19% of the load ## Concept Assign 10,000 users to 16 shards with a good hash (`md5(user_id) % 16`) and the per-shard **key counts** come out nearly equal — max/mean ≈ **1.06**, a textbook even spread. Now stop counting keys and start counting **requests**. Real traffic is Zipfian: a celebrity (`user0`) gets orders of magnitude more requests than a typical user. Sum *load* per shard instead of *keys* and the single shard holding `user0` carries **18.81%** of all requests — 3× its fair share — even though it holds ~1/16 of the keys. **Uniform over keys is not uniform over load.** The book's application-level fix is to split the hot key into 100 sub-keys (`user0#00`..`user0#99`) so its writes scatter across shards; do it and `user0`'s hot spot collapses from 18.81% to 4.73%. The twist you'll measure: max/mean only falls from 3.01 to 2.13 (Zipf has *many* hot keys, not one), and a **read** of `user0` must now gather all 100 sub-keys — fanning out to all 16 shards. ## Provenance *Designing Data-Intensive Applications*, 2nd ed. (Kleppmann & Riccomini, 2025), Chapter 7 — *Sharding*, §"Skewed Workloads and Relieving Hot Spots": > ...a hash function...can help reduce the hot spot. However...if one key is > known to be very hot...a simple technique is to add a random number to the > beginning or end of the key. Just a two-digit decimal random number would split > the writes to the key evenly across 100 different keys, allowing those keys to > be distributed to different shards. The book is careful about the cost: *"however, having split the writes across different keys, any reads now have to do additional work, as they have to read the data from all 100 keys and combine it."* This exercise makes both halves — the write spreading and the read amplification — into numbers you predict. ## Prerequisites The surprise is that two different distributions live on the same shards: keys are uniform, requests are not. These help; each line notes where it shows up. 1. **Hash uniformity** — a cryptographic hash (md5) maps arbitrary key strings to apparently-random buckets, so `key % 16` spreads *distinct keys* almost evenly. It shows up in Step A as key-count max/mean ≈ 1.06. 2. **Zipf / power-law popularity** — request frequency falls off as `1/rank^s` (here `s=1.1`), so the #1 key alone can be a large fraction of *all* traffic. It shows up in Step B as `user0` carrying 15.14% of requests by itself. 3. **Data skew vs. load skew** — balancing how many *keys* land on each shard (data) is a different problem from balancing how many *requests* each shard serves (load). A hash solves the first and can leave the second wide open. It shows up as the gap between Step A (max/mean 1.06) and Step B (max/mean 3.01). ### Where to learn the prerequisites - **Hash uniformity (#1):** DDIA §"Sharding by Hash of Key"; and Python's `hashlib` docs — any of md5/sha give a near-uniform spread for distinct inputs. - **Zipf / power laws (#2):** [Wikipedia, "Zipf's law"](https://en.wikipedia.org/wiki/Zipf%27s_law) — the `1/rank^s` head is where all the load concentrates. - **Data skew vs. load skew (#3):** DDIA §"Skewed Workloads and Relieving Hot Spots" — the section is *about* this distinction and why hashing does not close it. If only one is new, make it #3 — the whole exercise is the gap between an even *key* spread and a lopsided *load* spread on the very same shards. ## Environment Deterministic, so the numbers reproduce exactly: - macOS 26.5.2 (Darwin 25.5.0), arm64 - Python 3.15.0a8, standard library only — no Docker, no deps - Stable hashing: `int(hashlib.md5(s.encode()).hexdigest(), 16) % 16`, **not** Python's built-in `hash()` (which is salted per-process and would not reproduce) - Load computed analytically from the Zipf weights (no sampling), so there is zero Monte-Carlo noise — every shard percentage is exact every run ## Mental model One set of 16 shards, two distributions laid on top of it. - **Keys** — 10,000 user ids, each pinned to a shard by `md5(user_id) % 16`. The hash is (near-)uniform over distinct strings, so each shard gets ~625 keys. Counting keys, the shards look balanced. - **Load** — each user emits requests in proportion to a Zipf weight `1/(rank)^1.1`, with `user0` at rank 1 (the celebrity). A shard's load is the *sum of the weights of the users it holds*. Because one user's weight dwarfs the rest, the shard that happens to hold `user0` is hot regardless of how many keys it has. - **The fix** — replace the single key `user0` with 100 keys `user0#00`..`user0#99` and send 1/100 of its writes to each. Each sub-key hashes independently, so the concentrated weight scatters across shards. Reads must re-gather all 100. The whole thing is one deterministic script, `code/hot_key_skew.py` — it counts keys, then sums Zipf load, then splits the hot key and re-sums. ## Setup ``` cd study/ddia/ch07 python3 code/hot_key_skew.py ``` Do **not** read the output yet — make each prediction first. ## Steps ### Step 1 — key distribution **Predict.** 10,000 users are hashed onto 16 shards by `md5(user_id) % 16`. Counting *keys per shard*, how even is it? Give the max/mean imbalance you expect. **Observe.** ``` Step A -- 10000 keys over 16 shards, assigned by md5(user_id) % 16 per-shard KEY COUNT: shard 0 | 591 ## shard 1 | 636 ### shard 2 | 628 ### shard 3 | 616 ## shard 4 | 665 ### shard 5 | 662 ### shard 6 | 636 ### shard 7 | 599 ## shard 8 | 651 ### shard 9 | 607 ## shard 10 | 654 ### shard 11 | 603 ## shard 12 | 579 ## shard 13 | 590 ## shard 14 | 640 ### shard 15 | 643 ### min=579 max=665 mean=625.0 key-count imbalance max/mean = 1.06 (a good hash spreads keys evenly) ``` **max/mean = 1.06** — every shard holds within 6% of the mean 625. The hash did its job: distinct keys are spread almost perfectly evenly. If you stop here you'd conclude the shards are balanced. ### Step 2 — request load under a Zipf workload **Predict.** *Same shards, same assignment.* Now requests follow a Zipf law (`1/rank^1.1`), with `user0` the celebrity at rank 1. Sum *load* per shard. Will it still be ≈ 1.06? What fraction of all requests hits the single hottest shard? **Observe.** ``` Step B -- SAME shards, but requests follow Zipf (s=1.1); user0 = celebrity per-shard REQUEST LOAD (share of all requests): shard 0 | 8.04% ### shard 1 | 4.09% ## shard 2 | 5.03% ## shard 3 | 5.55% ## shard 4 | 3.29% # shard 5 | 5.02% ## shard 6 | 7.56% ### shard 7 | 2.98% # shard 8 | 8.79% #### shard 9 | 18.81% ######## shard 10 | 4.29% ## shard 11 | 4.10% ## shard 12 | 2.78% # shard 13 | 11.81% ##### shard 14 | 4.83% ## shard 15 | 3.02% # hottest is shard 9 (holds user0): 18.81% of ALL requests user0 alone is 15.14% of all requests load imbalance max/mean = 3.01 (uniform over keys is NOT uniform over load) ``` **max/mean = 3.01, not 1.06 — the same shards, a totally different picture.** Shard 9 serves **18.81%** of all requests (fair share is 6.25%), almost entirely because it holds `user0`, who alone is **15.14%** of traffic. The hash never had a chance: it balances *keys*, and load is glued to keys unequally. ### Step 3 — split the hot key **Predict.** Split `user0` into 100 sub-keys `user0#00`..`user0#99`, sending 1/100 of its writes to each and hashing each independently. Two questions: what does write-load max/mean drop to — all the way to ~1.0? — and how many of the 16 shards does a single *read* of `user0` now have to touch? **Observe.** ``` Step C -- split the hot key user0 into 100 sub-keys user0#00..user0#99 per-shard WRITE LOAD after splitting the hot key: shard 0 | 8.80% #### shard 1 | 4.55% ## shard 2 | 6.09% ## shard 3 | 6.16% ## shard 4 | 4.95% ## shard 5 | 6.08% ## shard 6 | 8.93% #### shard 7 | 3.74% # shard 8 | 9.25% #### shard 9 | 4.73% ## shard 10 | 5.19% ## shard 11 | 5.31% ## shard 12 | 3.38% # shard 13 | 13.32% ##### shard 14 | 5.89% ## shard 15 | 3.63% # user0's shard 9: 18.81% -> 4.73% (user0's hot spot is gone) but shard 13 -- the NEXT hot key's shard -- now leads at 13.32% write-load imbalance max/mean = 2.13 (was 3.01: better, not flat -- Zipf has MANY hot keys; you must split each one, not just the top) ...and the READ subtlety: a read of user0 must now gather all 100 sub-keys, which landed on 16 distinct shards -- so a single hot-key read now fans out to 16 of 16 shards. Write load spread; read load amplified. ``` **The fix works on `user0` — its shard falls from 18.81% to 4.73% — but max/mean only drops to 2.13, not ~1.0.** Splitting *one* key can only fix *one* hot spot; the next-hottest user (on shard 13) was already there under the celebrity and now stands revealed as the tallest bar. Zipf has a whole *head* of hot keys, so the technique has to be applied to each one you know about. And the price of the split is read amplification: the 100 sub-keys scattered across all 16 shards, so a read of `user0` now fans out to **16 of 16 shards**. ## What you should see - **Key counts** (Step A): max/mean = **1.06** — near-perfectly even. - **Request load** (Step B): max/mean = **3.01**; the hottest shard (9) serves **18.81%** of all requests; `user0` alone is **15.14%**. - **After splitting `user0`** (Step C): `user0`'s shard drops **18.81% → 4.73%**, but a new hot spot (shard 13) appears at **13.32%** and max/mean only falls to **2.13**. - A **read** of `user0` now touches **16 of 16** shards. ## Why A hash function is a promise about *identity*, not about *traffic*. `md5(user_id)` maps each distinct id to an apparently-random bucket, and with 10,000 distinct ids over 16 buckets the law of large numbers gives you ~625 per shard — max/mean 1.06. But the quantity you actually care about is *requests served*, and requests are attached to keys in wildly unequal amounts. Under Zipf, the weight of rank `r` is `1/r^1.1`; normalized over 10,000 users the rank-1 key `user0` carries **15.14%** of all traffic *by itself*. Hashing scatters that 15.14% onto whichever single shard `user0` lands on, and no amount of hash quality can divide one key's load across shards — a key is indivisible to the hash. So the shard holding `user0` sits at 18.81% (its own baseline ~3.7% of the tail plus `user0`'s 15.14%), while the mean is 6.25%: max/mean 3.01. **Uniform over keys ≠ uniform over requests** because the map from keys to load is itself skewed. The book's fix attacks the indivisibility directly: make the one hot key into 100 keys. Appending a two-digit suffix turns `user0` into `user0#00`..`user0#99`, and because each suffix hashes independently, 1/100 of `user0`'s writes go to each — its **write** load is divided by ~100 and scattered. That is why shard 9 collapses from 18.81% to 4.73%. But two things temper the win. First, splitting *one* key fixes *one* hot spot: the residual max/mean of 2.13 is the *next* celebrity (on shard 13), which Zipf guarantees exists — the head is many keys deep, so you split each hot key you know of, not just the top. Second, the split is a write-only trick. A **read** must reconstruct `user0` from all 100 sub-keys, and those 100 sub-keys landed on all **16** shards, so a hot-key read now fans out to every shard. You have traded read amplification for write spreading — a good trade only because writes to a celebrity vastly outnumber the (already-fanned-out) reads, and only worth doing for the handful of keys hot enough to matter. **The boundary — if the workload is uniform, the fix is pure overhead.** The whole problem exists because load per key is skewed. Flatten the popularity curve (every user equally likely) and Step B collapses back onto Step A: hashing already balances load, max/mean is ~1, and there is no hot key to split. Splitting keys under a uniform workload would only manufacture read fan-out for no write benefit. The technique is a *targeted* patch for the few keys the power law makes hot — it buys you nothing, and costs you reads, everywhere else. More generally: a hash gives you *data* balance for free; *load* balance is a separate, workload-dependent problem, and you pay for it per hot key. ### Go deeper 1. **Split the top-K.** Extend Step C to also split `user1`, `user2`, … and watch max/mean march toward ~1.0. How many of the head must you split before the residual imbalance is just hash noise (≈ 1.06)? 2. **Flatten the workload.** Set `ZIPF_S = 0` (every user equally likely) and re-run: confirm Step B's max/mean collapses to Step A's ~1.06 and the fix becomes pointless — the boundary condition, made concrete. 3. **Sharpen the celebrity.** Raise `ZIPF_S` to 1.5 or 2.0 and watch `user0`'s share climb past 50%. How high does the single hottest shard go, and how much of the imbalance does splitting just `user0` now remove?