Map–Reduce Model

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

# 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

# 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

# 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.