--- title: 'MapReduce Model' source: 'https://academia.sh/en/courses/distributed-data-processing/mapreduce-model' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Splits a computation into a map step that is independent per row and a reduce step per key.' - 'Measures, as moved rows and total work, that the map stays local and the reduce requires a shuffle.' - 'Computes how much local pre-reduction lowers the moved rows and the total work, and separates the operations that do not suit the method.' --- # MapReduce Model Splitting a computation into a step that works independently per row and a step that sums per key, and measuring that only one of the two steps touches the network: the map processes 2200 rows locally and moves 0 rows, the makespan is 611 and the total work stays at 2200, while the reduce step that produces the same result moves 1738 rows from the subscriber layout to the region key and raises the total work to 7414 and the makespan to 1914; when every worker reduces its own share before moving it, the moved rows are only 15 of 20 intermediate rows, the total work falls to 2245 and the makespan to 622, and the five region totals presented by the three setups remain exactly the same. In the previous two lessons every worker read its own share and processed its own share; no row passed from one worker to another. This was possible because the computations belonged to a special class: each row could be processed on its own. Computing the consumption difference needs two periods of a subscriber, but the same subscriber is already on a single worker; screening and deduplication also look at the row itself. When a result such as the total per region is wanted, this ends. The rows of a region are spread over two hundred subscribers, in a cluster partitioned by subscriber they have fallen on all four of the four workers, and they have to come together before they are summed. This lesson builds a model that splits the computation into two steps and asks a single thing: which step stays local, and which one touches the network. - **DC16.** The cluster, the pipeline and the two-step model are **fictional**; no real distributed processing framework, stream engine or cloud service is called. The seed is **20260218**. - **DC17.** **MapReduce** is the name of a model and does not denote a product. The map step applies a function to each row independently and produces **(key, value)** pairs, in this lesson one pair per row; the reduce step folds all the values of the same key into a single result. - **DC18.** The map step has **no dependency between rows**; so it can process the data where it is and does not touch the network. - **DC19.** The reduce step requires **all** the values of a key to be gathered on a single worker. If the data is not in that layout, a **shuffle** occurs, and every row that changes workers crosses the network. - **DC20.** The cost of a shuffled row is **`NETWORK_COST`, that is, 3**; this is a choice, and every makespan and total work measured in this lesson is sensitive to it. - **DC21.** The baseline is again **a single worker**: no shuffle, makespan **2250**, total work **2200**. - **DC22.** The rows enter the lesson in the layout they arrived from the previous lessons, that is, **partitioned by the `subscriber` key**. The size of the shuffle depends on this starting layout. - **DC23.** Makespan is the model's own unit and is not raw milliseconds; the numbers carry meaning only when compared with one another. ## The Two-Step Model The distinction the model draws comes from the computation itself. The question of whether one needs to look at other rows to do one row of a computation has two answers, and these two answers mean two steps. **Map** is the part that says no: it takes a key and a value out of the row and never looks at its neighbor. **Reduce** is the part that says yes: it wants all the values that belong to a key and cannot produce a single correct number without having them. The value of this distinction is that it says **in advance** which part of the computation costs network. No matter how many workers are added to the map step, no network traffic arises. The traffic of the reduce step, on the other hand, depends on the choice of the key and on the layout the data enters in. The block below builds the model with three functions: `map_rows` produces pairs, `local_reduce` folds a worker's own share, and `reduce_pairs` sums the result after the shuffle. ```python # FICTIONAL. The source, pipeline and cluster are defined in the lesson; no real distributed # processing framework, stream engine or cloud service is called. SEED, M32 = 20260218, 0xFFFFFFFF 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 REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped WORKERS, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50 def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets def shuffle(buckets, new_key, workers=WORKERS): """Repartition. A row that changes workers crosses the network.""" new = {i: [] for i in range(workers)} moved = 0 for old, records in buckets.items(): for k in records: target = trace_id(k[new_key]) % workers if target != old: moved += 1 new[target].append(k) return new, moved def cost(buckets, moved=0, workers=WORKERS): """Makespan = the busiest worker's work (parallel). Total work is counted separately.""" n = [len(v) for v in buckets.values()] return {"duration": max(n) + moved * NETWORK_COST // workers + TASK_OVERHEAD, "total_work": sum(n) + moved * NETWORK_COST, "moved": moved} # ---- FICTIONAL MapReduce: two steps def map_rows(share, key_fn, value_fn): """Produces one (key , value) pair per row. Local , never looks at a neighbor.""" return [{"key": key_fn(k), "value": value_fn(k)} for k in share] def local_reduce(pairs): """Sums the pairs in a worker's own share per key.""" acc = {} for c in pairs: acc[c["key"]] = acc.get(c["key"], 0) + c["value"] return [{"key": a, "value": v} for a, v in acc.items()] def reduce_pairs(buckets): """After the shuffle each worker sums its own keys , the driver merges.""" final = {} for share in buckets.values(): for c in local_reduce(share): final[c["key"]] = final.get(c["key"], 0) + c["value"] return final ``` ## The Map Stays Local The first measurement counts the map step alone. The rows enter the lesson partitioned by the `subscriber` key, and the map takes the region out of every row as the key and the consumption as the value. No worker looks at the row of another worker, so `shuffle` is never called and the moved rows are zero. ```python S = rows() K = distribute(S, "subscriber", WORKERS) MAPPED = {i: map_rows(share, lambda k: k["region"], lambda k: k["m3"]) for i, share in K.items()} print("single-worker baseline ->", cost(distribute(S, "subscriber", 1), 0, 1)) print("mapped pairs :", sum(len(v) for v in MAPPED.values())) print("map step ->", cost(K, 0)) ``` ``` single-worker baseline -> {'duration': 2250, 'total_work': 2200, 'moved': 0} mapped pairs : 2200 map step -> {'duration': 611, 'total_work': 2200, 'moved': 0} ``` Two thousand two hundred pairs come out of two thousand two hundred rows; the map does not change the number of rows, it changes the shape of the row. The total work is **2200**, as in the single-worker baseline, and four workers bring the makespan down from 2250 to **611**. This is the cleanest gain that will be seen in this course, and the reason is the definition of the map: because there is no dependency between rows, the computation is correct whatever the partition key is, and no row changes place. The practical conclusion from this is: how much of a computation fits into the map step tells how much of it will parallelize for free. Filtering, extracting a field, converting a unit and classifying per row belong to this step. Whatever does not belong is left to the second step. ## The Reduce Touches the Network The second step wants all the values of the same key. The region key is spread over all four of the four workers, so the worker that will compute the `merkez` (central) total holds only a quarter of the `merkez` pairs. The shuffle arises for this reason. The block below compares two reduce setups and checks the result of both against the single-worker baseline. ```python PRE_REDUCED = {i: local_reduce(c) for i, c in MAPPED.items()} BASELINE = reduce_pairs({0: map_rows(S, lambda k: k["region"], lambda k: k["m3"])}) print("intermediate rows after pre-reduction:", sum(len(v) for v in PRE_REDUCED.values())) print() print(f"{'setup':<28}{'moved':>7}{'makespan':>10}{'total work':>12}{'result':>17}") for name, source in (("reduce, direct", MAPPED), ("reduce, with pre-reduction", PRE_REDUCED)): y, moved = shuffle(source, "key", WORKERS) m = cost(K, moved) same = "same as baseline" if reduce_pairs(y) == BASELINE else "different" print(f"{name:<28}{moved:>7}{m['duration']:>10}{m['total_work']:>12}{same:>17}") print() print("baseline result:", BASELINE) ``` ``` intermediate rows after pre-reduction: 20 setup moved makespan total work result reduce, direct 1738 1914 7414 same as baseline reduce, with pre-reduction 15 622 2245 same as baseline baseline result: {'merkez': 14536, 'kuzey': 13338, 'dogu': 9695, 'bati': 9164, 'guney': 11721} ``` The first row writes the difference between the two steps. The same two thousand two hundred pairs move **0** rows in the map step and **1738** rows in the reduce step; 79% of the two thousand two hundred pairs change workers. The total work rises from **2200** to **7414**, and the makespan from 611 to **1914**. The 1738 rows that cross the network cost three units per row, that is, 5214 units, and this is the whole of the increase in total work. The number here does not come from the difficulty of the computation. Taking a region total is no heavier an operation than taking a consumption difference; what is heavy is that the data is partitioned by **the wrong key**. In the previous lesson it was measured that partitioning by the `region` key raised the skew to 3.32. There is no free key: partitioning by `subscriber` balances the map and forces the reduce into a shuffle, while partitioning by `region` zeroes the shuffle and leaves two workers idle. ## Pre-Reduction Produces the Same Result Cheaply The second row gives the real gain of the model. If every worker sums the pairs in its own share **before moving them**, the two thousand two hundred pairs coming out of the four workers fall to **20 intermediate rows**: each of the four workers produces one partial total for at most five regions. The shuffle now moves these twenty rows, and **15** of them change workers. The makespan becomes **622** and the total work **2245**; compared with the direct shuffle, the total work has fallen from **7414**, and a gap of nearly threefold has closed. The last column says the same thing in both setups. The five region totals are **exactly the same** as the single-worker baseline both in the direct shuffle and in the pre-reduction. This is the most unsettling observation of the course: the setup of 7414 units and the setup of 2245 units give the same output, and no check that looks at the output can tell them apart. What is wrong is not the computation but the cost; and the cost shows only if it is counted. Where the number twenty comes from also gives the limit of the gain. Because each of the four workers sees at most five distinct regions, pre-reduction could squeeze 2200 pairs into 20 rows; as the **number of distinct values** of the key grows, the squeeze decreases. The last block measures this over four reduce keys. ```python KEYS = (("region", lambda k: k["region"]), ("period", lambda k: k["period"]), ("region+period", lambda k: k["region"] + "|" + k["period"]), ("subscriber", lambda k: k["subscriber"])) print(f"{'reduce key':<14}{'distinct':>10}{'intermediate rows':>19}" f"{'direct':>8}{'pre-reduced':>13}{'total work':>12}") for name, f in KEYS: mapped = {i: map_rows(share, f, lambda k: k["m3"]) for i, share in K.items()} combined = {i: local_reduce(c) for i, c in mapped.items()} _, moved_direct = shuffle(mapped, "key", WORKERS) _, moved_pre = shuffle(combined, "key", WORKERS) print(f"{name:<14}{len(set(f(k) for k in S)):>10}{sum(len(v) for v in combined.values()):>19}" f"{moved_direct:>8}{moved_pre:>13}{cost(K, moved_pre)['total_work']:>12}") ``` ``` reduce key distinct intermediate rows direct pre-reduced total work region 5 20 1738 15 2245 period 11 44 1651 33 2299 region+period 55 220 1653 165 2695 subscriber 200 200 0 0 2200 ``` The fourth and the third column are two writings of the same computation, and the ratio between them closes together with the key. On the `region` key, which has five distinct values, the moved rows fall from 1738 to **15**; on the `region+period` key, which has fifty-five distinct values, they fall from 1653 to **165**, and the total work becomes **2695** instead of 2245. The squeeze ratio is determined by the number of intermediate rows, and that is the number of distinct values of the key multiplied by four workers. If the number of distinct values approaches the number of rows, pre-reduction cannot squeeze anything and pays the same cost as the direct shuffle. The last row is a separate case. When the reduce key is the same as the partition key, that is, when reducing to the `subscriber` key, the moved rows are **0** even in the direct writing: all the rows of every subscriber are already on a single worker. The shuffle is not a necessity but the cost of the **mismatch** between the reduce key and the layout the data enters in. The total work of this row is **2200**, the same as the map step itself. Pre-reduction has a condition, and not every computation meets it. The **order and grouping of the operation must be changeable**: summing is like this, because the sum of partial sums equals the sum of the whole. The smallest and the largest value are also like this. The median is not; the median of a worker's share cannot be computed and summed with the others, because the median wants to see all the values. A distinct count is not either: when the distinct subscriber counts in the shares of two workers are summed, the common subscribers are counted twice. For these computations either all the rows are moved or an approximate result is accepted. The model does not make every computation cheaper; it **says** which ones it makes cheaper. ## Summary - **MapReduce** splits a computation into two steps: the **map** is applied to each row independently and produces a (key, value) pair, the **reduce** folds all the values of a key into a single result. - The map step is local. 2200 rows produce 2200 pairs, the moved rows are **0**, the total work stays at **2200** and four workers bring the makespan from 2250 down to **611**. - The reduce step touches the network. In passing from the `subscriber` layout to the `region` key, **1738** rows change workers; the total work rises to **7414** and the makespan to **1914**. The whole of the increase is the three-unit cost of the moved rows. - When every worker reduces its own share before moving it, 2200 pairs fall to **20 intermediate rows**, **15** of them are moved, and the same result is produced for a total work of **2245** and a makespan of **622**. - The five region totals presented by the three setups are exactly the same. A check that looks at the output **cannot tell apart** the setup of 7414 units and the one of 2245; the difference is only in the cost. - Pre-reduction is valid only for operations whose order and grouping can be changed. Sum, minimum and maximum are suitable; median and distinct count are not. ## Next Step In three lessons the job was split, placed and run. In every measurement the whole computation ran: all two thousand two hundred rows were mapped, all were moved, all were reduced. Yet in a case where a report asks for only one region or the last three periods, most of these rows should never have been touched, and which computation is **really needed** was never asked. How the workers are shared across the cluster was not asked either; four workers were assumed ready in every measurement. The next two lessons measure these two questions: how many rows a computation that is not used in the result leaves untouched, and where the fixed cost of increasing the number of workers eats the gain.