--- title: 'Map–Reduce Model' source: 'https://academia.sh/en/courses/visualization-reporting-and-analysis-at-scale/mapreduce-model' course: 'Visualization, Reporting and Analysis at Scale' language: en updated: '2026-10-07T04:31:14+00:00' license: 'CC BY-SA 4.0' objectives: - 'Writes the map, shuffle, and reduce functions and measures the number of records moved in the shuffle' - 'Compares how many times local reduction cuts the transfer across four grouping keys and writes what the gain depends on' - 'Puts side by side the records moved and the distinct numbers produced by a computation that needs two shuffles' - 'Tests whether the reduce function is associative by changing the part count and separates out the computation that does not enter the model' --- # Map–Reduce Model The simplest abstraction of distributed computation is built and run with three functions, and the measure becomes the number of records moved in the shuffle: on the fictional dataset spread across eight source parts, a regional average moves 3,199 records in the order that shuffles first and then reduces and 40 records in the order that reduces locally first and then shuffles, a gain of 79.97 times. The gain depends on the number of keys and falls to 2.53 times on the subscriber key. The same question gives two values with two keys: the east average is 24.674 m³ weighted by rows and 24.814 m³ as the average of subscriber averages, and the records moved by the two passes rise from 40 to 1,305. If the reduce function is not associative the computation does not enter the model: the sum and the maximum do not change across seven different part counts, the median of part medians gives seven different values, and its spread is 0.2175 m³. The previous lesson wrote the point at which the dataset no longer fits on a single machine as a row count: 22,525,682 rows under the chosen decision. Past that point the computation has to be divided, but "divided" is not a recipe by itself. What is divided, what travels between the parts, and how many records travel. This lesson builds and runs the simplest abstraction of distributed computation. The abstraction is three functions: **map** turns a row into a key and an intermediate value, the **shuffle** moves the intermediate values to the reducers by key, and **reduce** closes the intermediate values of the same key into a single value. The measure of the lesson is in the middle step: how many records are moved in the shuffle. The second number next to it is how much of the reduction was done before moving the record. - **AS7.** The dataset is inherited from the Data Collection and Preparation, Data Analysis with Python, and Statistical Analysis and Exploratory Data Analysis courses and **is fictional**: 3,199 rows, 1,329 subscribers, three periods, five regions, tariff tiers of 10/25/40 m³. The seed is 20260218 and the output is the same from run to run. **It is a model.** - **AS8.** No distributed processing engine is installed; the source is taken to be **spread across eight parts** and the three functions are written with the standard library. **It is a model.** - **AS9.** A record moved in the shuffle is a (key, intermediate value) pair sent to the reducers. Raw milliseconds are not measured. - **AS10.** The intermediate value is a decision. For the average the intermediate value is not a single number but a **(total, count)** pair; this choice makes the reduce function associative. - **AS11.** Whether a computation enters the model depends on the reduce function being **associative and commutative** over partial results. The test is done by running: the same computation is run with seven different part counts and the number of distinct numbers that come out is counted. - **AS12.** The part boundary is drawn by row count; the parts are in the order the source gives. ## Three Functions ```python # map_reduce.py — the FICTIONAL dataset of the regional measurement network and three functions: # map, shuffle, reduce. MODEL: no distributed engine is installed; the source is taken to be # spread across PARTS parts and the records moved in the shuffle are counted. import math SEED, RAW, M32, PARTS = 20260218, 1400, 0xFFFFFFFF, 8 REGION = [("north", 0.28, 21), ("south", 0.22, 17), ("east", 0.18, 26), ("west", 0.14, 14), ("center", 0.18, 23)] def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value def pick(u, w): t = 0.0 for i, x in enumerate(w): t += x if u < t: return i return len(w) - 1 def tier(v): return "0-10" if v <= 10 else "10-25" if v <= 25 else "25-40" if v <= 40 else "over 40" DATASET = [] for i in range(RAW): r = rng(SEED + i) b = REGION[pick(r(), [x[1] for x in REGION])] household = 1 + pick(r(), [0.06, 0.24, 0.30, 0.24, 0.11, 0.05]) satisfied = 1 + pick(r(), [0.08, 0.14, 0.27, 0.34, 0.17]) if r() < 0.046: # meter that failed to register continue sub_id, q = 10001 + i, rng(SEED + 7000 + 10001 + i) for d in range(pick(q(), [0.05, 0.12, 0.21, 0.62])): v = 0.0 if q() < 0.038 else math.floor( b[2] * math.exp((q() + q() + q() - 1.5) * 0.62) * (1 - d * 0.05) * 100 + 0.5) / 100 DATASET.append((sub_id, b[0], d + 1, household, satisfied, v, tier(v))) q() def split_parts(dataset, p): n = len(dataset) return [dataset[i * n // p:(i + 1) * n // p] for i in range(p)] def map_row(row, key_fn): return (key_fn(row), (row[5], 1)) # intermediate value: (total, count) def combine(a, b): return (a[0] + b[0], a[1] + b[1]) def run(parts, key_fn, use_combiner): """Map -> (local reduce, if any) -> shuffle -> reduce. The second number returned is the number of records moved in the shuffle.""" moved = [] for p in parts: pairs = [map_row(s, key_fn) for s in p] if use_combiner: local = {} for k, v in pairs: local[k] = combine(local[k], v) if k in local else v pairs = sorted(local.items(), key=lambda x: str(x[0])) moved += pairs bucket = {} for k, v in moved: bucket[k] = combine(bucket[k], v) if k in bucket else v return {k: (v[0] / v[1], v[1]) for k, v in bucket.items()}, len(moved) P = split_parts(DATASET, PARTS) result, moved = run(P, lambda s: s[1], False) print(f"seed {SEED}; {len(DATASET)} rows, {PARTS} source parts " f"{[len(x) for x in P]}") print(f"{'region':<10}{'rows':>7}{'average m3':>14}") for b in sorted(result): print(f"{b:<10}{result[b][1]:>7}{result[b][0]:>14.3f}") print(f"east - north = {result['east'][0] - result['north'][0]:.3f} m3; " f"records moved in the shuffle {moved}") ``` ``` seed 20260218; 3199 rows, 8 source parts [399, 400, 400, 400, 400, 400, 400, 400] region rows average m3 center 600 22.396 east 557 24.674 north 921 20.210 south 645 16.602 west 476 13.778 east - north = 4.463 m3; records moved in the shuffle 3199 ``` The dataset was spread across eight parts and was never brought together as a whole anywhere, yet the anchor of the course did not change: the east average is 24.674 m³, the north 20.210 m³, the difference 4.463 m³. When the model is built correctly, dividing does not move the answer; what it moves is the records carried to produce the answer. In this first run the records moved are 3,199, that is, the whole dataset. Map produced one pair for every row and all of the pairs went to the reducers. Reduce then ran and gave five numbers. For five numbers, 3,199 records were moved. What the intermediate value is turns out to be an invisible decision at this point. The map function does not send the reading alone; it puts a 1 next to it and sends a **(total, count)** pair. If a single number were sent, the reduce function would have to take the average of averages, and as soon as the part sizes were unequal the result would drift. Sending a pair does not double the records moved — the number of records is the same, the inside of the record widens — but it makes the reduce function associative. The third section of the lesson measures the case where this property is absent. ## Records Moved in the Shuffle ```python # transfer.py — the same aggregation is run in two orders: the order that shuffles first and # then reduces, and the order that reduces locally first and then shuffles. The measure is the # number of records moved in the shuffle. KEYS = (("region", lambda s: s[1]), ("region + period", lambda s: (s[1], s[2])), ("region + tier", lambda s: (s[1], s[6])), ("subscriber_id", lambda s: s[0])) print(f"{'grouping key':<18}{'keys':>9}{'shuffle first':>15}" f"{'reduce first':>14}{'gain':>9}{'same result':>13}") for label, f in KEYS: a, ta = run(P, f, False) b, tb = run(P, f, True) same = all(abs(a[k][0] - b[k][0]) < 1e-9 for k in a) print(f"{label:<18}{len(a):>9}{ta:>15}{tb:>14}{ta / tb:>8.2f}x" f"{('yes' if same else 'no'):>13}") print(f"records sent by the combiner are the distinct key count per part; " f"{PARTS} parts and 5 regions make {PARTS * 5} records") ``` ``` grouping key keys shuffle first reduce first gain same result region 5 3199 40 79.97x yes region + period 15 3199 120 26.66x yes region + tier 18 3199 137 23.35x yes subscriber_id 1260 3199 1265 2.53x yes records sent by the combiner are the distinct key count per part; 8 parts and 5 regions make 40 records ``` The order decision does not move the result, it moves the transfer. In all four rows the last column says "yes": shuffling first and reducing afterward and reducing locally first and shuffling afterward give the same five numbers. What changes is the records that pass through the shuffle, and on the region key they fall from 3,199 to 40 — **79.97 times**. Where the forty comes from is a one-line product: eight parts, five regions in each part. Local reduction brings all the rows in a part down to the number of keys in that part; the records moved now come not from the size of the dataset but from **the number of parts times the number of distinct keys within a part**. Even if the dataset grew a thousand times, the records moved on the region key would stay at forty. The last row draws the limit. The subscriber key has 1,260 distinct keys and the row count is 3,199; a subscriber has at most three rows. Local reduction brings two and a half rows per part down to one on average and the gain falls to 2.53 times. The combiner's gain comes from **the ratio of the number of keys to the number of rows**; as the number of keys approaches the number of rows, the combiner finds nothing to carry. The two middle rows show how this ratio works. When period is added to region, the number of keys rises from five to fifteen and the records moved come to exactly eight times fifteen, that is, 120: all fifteen keys are present in all eight parts. When tier is added, the number of keys is eighteen, but the records moved are 137, not 144. The shortfall of seven records is not a gain but a sparsity: two of the eighteen combinations of region and tier hold only a few rows in the whole dataset and do not appear at all in seven of the eight parts. The records moved are the sum of **the number of keys actually seen per part**, not of the number of keys, and this number is known only by running. ## Two Passes and the Computation That Does Not Enter the Model ```python # passes.py — some computations do not fit a single shuffle, and some do not fit at all. # First the two-pass computation, then whether the reduce function is associative is tested: # the same computation is run with seven different part counts. import statistics subs, t1 = run(P, lambda s: s[0], True) # pass 1: key is subscriber REGION_OF = {s[0]: s[1] for s in DATASET} second = [(REGION_OF[k], (v[0], 1)) for k, v in subs.items()] P2 = split_parts(second, PARTS) # pass 2: key is region t2 = sum(len({k for k, _ in p}) for p in P2) mean2 = {} for p in P2: for k, v in p: mean2[k] = combine(mean2[k], v) if k in mean2 else v single, t_single = run(P, lambda s: s[1], True) print(f"{'region':<10}{'single pass: row-weighted':>28}" f"{'two passes: subscriber average':>32}{'difference':>12}") for b in sorted(single): a2 = mean2[b][0] / mean2[b][1] print(f"{b:<10}{single[b][0]:>28.3f}{a2:>32.3f}{a2 - single[b][0]:>12.3f}") print(f"single pass moves {t_single} records; two passes move {t1} + {t2} = {t1 + t2} records") medians = [] print(f"{'parts':<8}{'total m3':>12}{'largest':>10}" f"{'median of part medians':>32}") for p in (2, 3, 4, 5, 8, 16, 32): splits = split_parts(DATASET, p) m = statistics.median([statistics.median([x[5] for x in y]) for y in splits]) medians.append(m) print(f"{p:<8}{sum(sum(x[5] for x in y) for y in splits):>12.2f}" f"{max(max(x[5] for x in y) for y in splits):>10}{m:>32.4f}") bucket_count = sum(len({x[5] for x in p}) for p in P) print(f"true median {statistics.median([s[5] for s in DATASET])} m3; seven part " f"counts give {len(set(medians))} distinct values, spread " f"{max(medians) - min(medians):.4f} m3") print(f"readings have two decimals: {len({s[5] for s in DATASET})} distinct values; count " f"combiner moves {bucket_count} records, gain {len(DATASET) / bucket_count:.2f}x") ``` ``` region single pass: row-weighted two passes: subscriber average difference center 22.396 22.483 0.087 east 24.674 24.814 0.140 north 20.210 20.351 0.141 south 16.602 16.728 0.127 west 13.778 13.935 0.158 single pass moves 40 records; two passes move 1265 + 40 = 1305 records parts total m3 largest median of part medians 2 63060.63 52.47 18.9900 3 63060.63 52.47 18.9700 4 63060.63 52.47 18.9250 5 63060.63 52.47 18.8400 8 63060.63 52.47 18.9725 16 63060.63 52.47 18.8900 32 63060.63 52.47 19.0575 true median 18.98 m3; seven part counts give 7 distinct values, spread 0.2175 m3 readings have two decimals: 1788 distinct values; count combiner moves 2839 records, gain 1.13x ``` The first table is the hidden decision of this lesson. "Regional average" looks like a single question, yet it is two separate computations. In the single pass the key is the region and every row carries equal weight: east 24.674 m³. In the two passes the reduction is first by the subscriber key, then the subscriber results are reduced by the region key, and every subscriber carries equal weight: east 24.814 m³. The difference is 0.140 m³ and in the same direction in all five regions. Both are defensible; when which one was chosen is not written down, the number in the report is not a finding but a choice. The cost is in the table too. The single pass moves 40 records, the two passes 1,265 plus 40, a total of 1,305 records — **thirty-three times**. As the number of passes grows, the records moved multiply, because the output of the intermediate pass becomes the input of the next pass and that output is as large as the number of keys. The second table draws the limit of entering the model. The sum and the maximum do not budge across seven different part counts: 63,060.63 m³ and 52.47 m³. The reduce functions are associative and commutative over partial results, so the splitting stays invisible. The median, on the other hand, gives **seven distinct values** across the seven part counts; the spread is 0.2175 m³ and none of them equals the true median, 18.98 m³. The median of part medians is not even an approximation; it is a number that depends on the part count. The last line gives the price of the way out. When the intermediate value of the median is a **count** instead of the median, it enters the model, but it does not save the day on this dataset: because the readings are written with two decimals there are 1,788 distinct values, and the count combiner moves 2,839 records, that is, a gain of 1.13 times. The number of decimals is a display decision and here it determines the length of the shuffle: the same readings written with one decimal would make the combiner work. ## Summary - The three functions ran and, although the dataset was spread across eight parts, the anchor did not change: east 24.674 m³, north 20.210 m³, difference 4.463 m³. - The order decision moves the transfer, not the result: on the region key shuffling first moves 3,199 records, reducing locally first moves 40, and the gain is 79.97 times. - The records a combiner moves are the product of the part count and the number of distinct keys in a part; on the subscriber key, 1,260 keys cut the gain to 2.53 times. - The same question gives two values with two keys: the east average is 24.674 m³ weighted by rows and 24.814 m³ as the average of subscriber averages, and the two passes move 1,305 records instead of 40. - A computation does not enter the model if the reduce function is not associative: the sum and the maximum come out the same across seven part counts, the median of part medians gives seven different values, and its spread is 0.2175 m³. - When the intermediate value of the median is turned into a count it enters the model, but with readings of two decimals there are 1,788 distinct values and the gain falls to 1.13 times. ## Next Step This lesson counted the records moved and showed that the transfer depends on the order, but it never asked who runs the three functions. In a real job the parts are defined somewhere, tasks are distributed, a failed task is retried, and the results are collected; none of these processes rows and all of them are added per job. The next lesson counts the **coordination overhead item** of data frame processing on a cluster: into how many stages a chain of operations is split, how many items are added per stage, how many a retry raises this number to, and at which job size the number of items exceeds the number of rows processed.