#!/usr/bin/env python3 """A broadcast join moves orders of magnitude less data than a reduce-side join, DDIA 2e Ch. 11. Join a large FACT table (1,000,000 activity events keyed by user_id) to a small DIMENSION table (1,000 user profiles) with a tiny in-process MapReduce, two ways: Reduce-side / sort-merge join -- every record of BOTH tables is emitted keyed by user_id and shuffled across the network to reducers, which group by key and join. Bytes over the network = size(fact) + size(dim): the WHOLE big table moves. Broadcast / map-side (hash) join -- the small dim table is shipped to every mapper ONCE; each mapper builds a local hash of it and joins its slice of the fact table in place. The fact table NEVER crosses the shuffle. Bytes over the network = N_mappers * size(dim): near-zero when dim << fact. We instrument the shuffle with a real byte counter (the sum of the serialized sizes of the records that cross the shuffle boundary) and prove both strategies emit the IDENTICAL join result -- same row count, same order-independent checksum. Pure standard library, seeded -> the byte counts reproduce exactly. """ import hashlib import json import random SEED = 42 N_FACT = 1_000_000 # activity events in the big fact table N_DIM = 1_000 # user profiles in the small dimension table N_REDUCERS = 16 # reducers the reduce-side join shuffles into N_MAPPERS = 8 # mappers the broadcast join replicates the dim to MOD = (1 << 61) - 1 # prime modulus for the order-independent checksum EVENTS = ("click", "view", "purchase", "scroll", "hover", "share") COUNTRIES = ("US", "GB", "DE", "IN", "BR", "JP", "FR", "CA") PLANS = ("free", "pro", "team", "enterprise") def sizeof(record): """Serialized wire size of one record, in bytes -- what a shuffle would move.""" return len(json.dumps(record, separators=(",", ":")).encode("utf-8")) def build_tables(): """Deterministically generate the fact and dimension tables.""" rnd = random.Random(SEED) fact = [] for i in range(N_FACT): fact.append({ "user_id": rnd.randint(0, N_DIM - 1), "event": rnd.choice(EVENTS), "item_id": rnd.randint(10000, 99999), "ts": 1690000000 + i, }) dim = [] for uid in range(N_DIM): dim.append({ "user_id": uid, "name": f"user_{uid:04d}", "country": rnd.choice(COUNTRIES), "plan": rnd.choice(PLANS), }) return fact, dim def row_hash(frec, drec): """Order-independent fingerprint of one joined output row.""" canonical = json.dumps( [frec["user_id"], frec["event"], frec["item_id"], frec["ts"], drec["name"], drec["country"], drec["plan"]], separators=(",", ":"), ) return int.from_bytes(hashlib.md5(canonical.encode("utf-8")).digest()[:8], "big") # --- Reduce-side (sort-merge) join: shuffle BOTH tables to reducers by key --- def reduce_side_join(fact, dim): shuffle_bytes = 0 inboxes = [[] for _ in range(N_REDUCERS)] # one inbox per reducer # Map phase: emit every record keyed by user_id, tagged with its source. # EVERY emitted record crosses the shuffle boundary to reach its reducer. for row in fact: shuffle_bytes += sizeof(row) inboxes[row["user_id"] % N_REDUCERS].append(("F", row)) for row in dim: shuffle_bytes += sizeof(row) inboxes[row["user_id"] % N_REDUCERS].append(("D", row)) # Reduce phase: group each inbox by user_id, join the dim record to its facts. out_count = 0 checksum = 0 for inbox in inboxes: groups = {} for tag, rec in inbox: g = groups.setdefault(rec["user_id"], {"F": [], "D": None}) if tag == "D": g["D"] = rec else: g["F"].append(rec) for g in groups.values(): if g["D"] is not None: for frec in g["F"]: out_count += 1 checksum = (checksum + row_hash(frec, g["D"])) % MOD return shuffle_bytes, out_count, checksum # --- Broadcast (map-side hash) join: replicate the small dim to every mapper --- def broadcast_join(fact, dim): # Broadcast phase: the dim table is copied to every mapper ONCE. That, and # only that, crosses the network. The fact table stays where it already lives. dim_bytes = sum(sizeof(r) for r in dim) shuffle_bytes = N_MAPPERS * dim_bytes # Map phase: each mapper builds a local hash of the dim and joins its slice # of the fact table in place. (One shared dict models the per-mapper hash.) dim_index = {r["user_id"]: r for r in dim} out_count = 0 checksum = 0 for frec in fact: drec = dim_index.get(frec["user_id"]) if drec is not None: out_count += 1 checksum = (checksum + row_hash(frec, drec)) % MOD return shuffle_bytes, out_count, checksum def human(nbytes): for unit in ("B", "KB", "MB", "GB"): if nbytes < 1024 or unit == "GB": return f"{nbytes:.1f} {unit}" if unit != "B" else f"{nbytes} B" nbytes /= 1024 def main(): fact, dim = build_tables() fact_bytes = sum(sizeof(r) for r in fact) dim_bytes = sum(sizeof(r) for r in dim) print("Join a FACT table to a DIMENSION table with an in-process MapReduce.") print(f" fact: {N_FACT:,} activity events = {human(fact_bytes)} on disk") print(f" dim: {N_DIM:,} user profiles = {human(dim_bytes)} on disk") print(f" the fact table is {fact_bytes / dim_bytes:,.0f}x larger than the dim table\n") r_bytes, r_rows, r_sum = reduce_side_join(fact, dim) print("reduce-side (sort-merge) join -- shuffle BOTH tables to reducers by user_id:") print(f" rows joined = {r_rows:,}") print(f" bytes shuffled = {human(r_bytes)} (= size(fact) + size(dim))\n") b_bytes, b_rows, b_sum = broadcast_join(fact, dim) print(f"broadcast (map-side hash) join -- ship the dim to all {N_MAPPERS} mappers, fact stays put:") print(f" rows joined = {b_rows:,}") print(f" bytes shuffled = {human(b_bytes)} (= {N_MAPPERS} mappers x size(dim))\n") same = (r_rows == b_rows) and (r_sum == b_sum) print("same join result?") print(f" reduce-side: {r_rows:,} rows, checksum {r_sum}") print(f" broadcast: {b_rows:,} rows, checksum {b_sum}") print(f" IDENTICAL -> {same}\n") print("data moved over the network:") print(f" reduce-side {human(r_bytes)}") print(f" broadcast {human(b_bytes)}") print(f" broadcast moves {r_bytes / b_bytes:,.0f}x LESS data for the SAME answer.") if __name__ == "__main__": main()