MapReduce Model

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.

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

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.

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.

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.