#!/usr/bin/env python3 """One hot key sends most of a skewed reduce-side join to a single reducer. An in-process MapReduce shuffle: every fact record is keyed by user_id and sent to reducer `md5(key) % R`, where the matching reducer joins it against the dimension row for that key. Wall-clock is modeled as the STRAGGLER -- the reducer with the most records -- because every reducer runs in parallel and the job is not done until the slowest one is. - Uniform keys: the R reducers get balanced load; doubling R halves the max load -> near-linear speedup. - One hot key ("celebrity", ~85% of records): all its records hash to ONE reducer, which then holds ~85% of the whole job. Every other reducer idles. - Adding reducers does NOT help: the hot key is atomic under hashing (it always lands on the same single reducer), so the straggler's absolute load barely moves as R grows. - Salting: replace `celebrity` with `celebrity#0..celebrity#(S-1)` on the fact side and replicate the dim row across all S salts. The hot key becomes S sub-keys that hash independently, so its load spreads across reducers and the max load collapses. Deterministic: random.Random(42) for the key assignment, and a STABLE md5-based partitioner (never Python's per-process-salted hash()), so every count below reproduces exactly. """ import hashlib import random N_FACT = 1_000_000 # activity events (the large "fact" side) HOT_KEY = "celebrity" # one dominant user HOT_FRACTION = 0.85 # ~85% of all events belong to the celebrity N_COLD_KEYS = 10_000 # the long tail: ordinary users SEED = 42 # ---------------------------------------------------------------- shuffle core def partition(key, R): """Stable hash partitioner: which reducer file does this key's records go to? md5 (not Python's hash(), which is salted per process) so the assignment is identical on every run and every machine. """ return int(hashlib.md5(key.encode()).hexdigest(), 16) % R def reducer_loads(keys, R): """Count records landing on each of R reducers under the md5 partitioner.""" loads = [0] * R for k in keys: loads[partition(k, R)] += 1 return loads def summarize(loads): total = sum(loads) mx = max(loads) mean = total / len(loads) hot_reducer = loads.index(mx) return { "R": len(loads), "total": total, "max": mx, "mean": mean, "max_over_mean": mx / mean, "hot_reducer": hot_reducer, "hot_share": mx / total, } def histogram(loads, width=40): """Render per-reducer record counts as a bar chart (for small R).""" mx = max(loads) lines = [] for i, n in enumerate(loads): bar = "#" * max(1, round(width * n / mx)) lines.append(f" reducer {i:2d} | {n:>8,} {bar}") return "\n".join(lines) # ---------------------------------------------------------------- workloads def make_uniform_keys(rng): """Every event belongs to a uniformly-random ordinary user -- no hot key.""" return [f"user{rng.randrange(N_COLD_KEYS)}" for _ in range(N_FACT)] def make_skewed_keys(rng): """~85% of events belong to `celebrity`; the rest spread over the long tail.""" keys = [] for _ in range(N_FACT): if rng.random() < HOT_FRACTION: keys.append(HOT_KEY) else: keys.append(f"user{rng.randrange(N_COLD_KEYS)}") return keys def salt_hot_key(keys, rng, S): """Fact-side fix: rewrite each `celebrity` record to `celebrity#{0..S-1}`. Each salted sub-key hashes independently, so the celebrity's records fan out across reducers instead of piling onto one. The dim side must replicate the celebrity's profile row across all S salts so the join still matches -- that replication is the cost of the fix (reported below). """ out = [] for k in keys: if k == HOT_KEY: out.append(f"{HOT_KEY}#{rng.randrange(S)}") else: out.append(k) return out # ---------------------------------------------------------------- report def main(): line = "=" * 68 # ---- Step 1: uniform keys scale with R ---------------------------------- rng = random.Random(SEED) uni = make_uniform_keys(rng) print(line) print("STEP 1 -- UNIFORM keys: reduce-side join, partition by md5(key) % R") print(line) R = 8 loads = reducer_loads(uni, R) s = summarize(loads) print(f"{N_FACT:,} events, {N_COLD_KEYS:,} distinct users, no hot key. R = {R} reducers:") print(histogram(loads)) print(f" max={s['max']:,} mean={s['mean']:,.0f} " f"max/mean = {s['max_over_mean']:.2f} (balanced: the hash spreads load evenly)") print() print("Same uniform workload, more reducers -- wall-clock = the straggler (max load):") print(f" {'R':>4} | {'max reducer load':>16} | {'max/mean':>9} | {'wall-clock vs R=8':>18}") base = None for Rn in (8, 16, 64): sn = summarize(reducer_loads(uni, Rn)) if base is None: base = sn["max"] print(f" {Rn:>4} | {sn['max']:>16,} | {sn['max_over_mean']:>9.2f} | " f"{base / sn['max']:>16.2f}x") print(" -> doubling R roughly halves the max load: near-linear speedup.") print() # ---- Step 2: one hot key floods one reducer ----------------------------- rng = random.Random(SEED) skew = make_skewed_keys(rng) n_hot = sum(1 for k in skew if k == HOT_KEY) print(line) print("STEP 2 -- SKEWED keys: one celebrity owns ~85% of the events") print(line) R = 8 loads = reducer_loads(skew, R) s = summarize(loads) print(f"{N_FACT:,} events, but `{HOT_KEY}` appears in {n_hot:,} of them " f"({n_hot / N_FACT:.1%}). R = {R} reducers:") print(histogram(loads)) print(f" hottest is reducer {s['hot_reducer']} (holds `{HOT_KEY}`): " f"{s['max']:,} records = {s['hot_share']:.1%} of the whole job") print(f" max={s['max']:,} mean={s['mean']:,.0f} " f"max/mean = {s['max_over_mean']:.2f} (one reducer holds the wall-clock hostage)") print(f" the other {R - 1} reducers finish early and idle while it runs the job alone.") print() # ---- Step 3: more reducers do NOT help ---------------------------------- print(line) print("STEP 3 -- add reducers to the SKEWED job: does the straggler shrink?") print(line) print(f" {'R':>4} | {'hot reducer load':>16} | {'hot share':>9} | {'max/mean':>9} | {'wall-clock vs R=8':>18}") base = None for Rn in (8, 16, 64): sn = summarize(reducer_loads(skew, Rn)) if base is None: base = sn["max"] print(f" {Rn:>4} | {sn['max']:>16,} | {sn['hot_share']:>8.1%} | " f"{sn['max_over_mean']:>9.2f} | {base / sn['max']:>16.2f}x") print(f" -> the hot reducer's ABSOLUTE load barely moves (~{n_hot:,} celebrity") print(" records always hash to ONE reducer). max/mean rises only because the") print(" mean shrinks. 8x the reducers buys ~1x the speedup: no help at all.") print() # ---- Step 4: salt the hot key ------------------------------------------ S = 16 rng = random.Random(SEED) salted = salt_hot_key(skew, rng, S) print(line) print(f"STEP 4 -- SALT the hot key: `{HOT_KEY}` -> `{HOT_KEY}#0..#{S - 1}` " f"(S = {S}), replicate dim") print(line) R = 8 before = summarize(reducer_loads(skew, R)) loads = reducer_loads(salted, R) s = summarize(loads) print(f"Same {N_FACT:,} events, same R = {R} reducers, hot key split into {S} salts:") print(histogram(loads)) print(f" max={s['max']:,} mean={s['mean']:,.0f} " f"max/mean = {s['max_over_mean']:.2f} (was {before['max_over_mean']:.2f})") print(f" straggler load {before['max']:,} -> {s['max']:,} " f"({before['max'] / s['max']:.1f}x lighter)") print(f" hot-key share of the busiest reducer " f"{before['hot_share']:.1%} -> {s['hot_share']:.1%}") salts_used = len({partition(f"{HOT_KEY}#{i}", R) for i in range(S)}) print(f" the {S} salts landed on {salts_used} of {R} reducers, so the celebrity's") print(f" load now spreads instead of piling up. COST: the dim row for `{HOT_KEY}`") print(f" is replicated across all {S} salts (a read/recombine must re-union them).") print() print(line) print("SUMMARY") print(line) print(f" uniform, R 8->64: max load {summarize(reducer_loads(uni,8))['max']:,} " f"-> {summarize(reducer_loads(uni,64))['max']:,} (8x reducers, ~8x faster)") print(f" skewed, R 8->64: max load {summarize(reducer_loads(skew,8))['max']:,} " f"-> {summarize(reducer_loads(skew,64))['max']:,} (8x reducers, ~1x faster)") print(f" salted, R = 8: max load {before['max']:,} -> {s['max']:,} " f"({before['max'] / s['max']:.1f}x faster, same 8 reducers)") if __name__ == "__main__": main()