P06 hands-on — MapReduce framework, block by block

Why a restricted programming model is what makes fault tolerance possible.

Source: handson/h06_mapreduce.py --- run it with python3 handson/h06_mapreduce.py
Full project spec: P06 — MapReduce-Style Framework

MapReduce is usually taught as a way to process large data. It is more usefully understood as a bargain: you give up arbitrary computation, and in exchange the framework may re-run any task, anywhere, at any time, without asking you.

This file makes that bargain concrete. The assembly runs the same job with three different worker-failure schedules and gets byte-identical output every time, which is possible only because map and reduce are pure functions of their input. The atomic-rename commit is what extends that guarantee to the outside world: duplicate task attempts produce one file, so at-least-once execution becomes exactly-once effect.

The straggler simulation at the end is the part most implementations skip, and it is where the wall-clock time actually goes.

Contents

• Block 1 — Input splitting
• Block 2 — map + partition
• Block 3 — Shuffle: the all-to-all
• Block 4 — Combiner
• Block 5 — Reduce + atomic commit
• Block 6 — Stragglers and backup tasks
• The assembly
• The design space
• The shuffle is the system
• Stragglers, and the arithmetic of maxima
• Advanced algorithms and alternatives
• Hardware: what changed since 2004
• How this connects to the rest of the track
• Failure modes at scale
• Primary sources
• Running it
• What to do with this

How to read this page

Each block below is a self-contained lego piece: it builds one mechanism, proves it works on its own, and returns what the next block needs. The code is the real source, sliced out of the script. The output underneath it is the real output, captured by running that script --- not transcribed, not idealised. Where a measurement contradicted what I expected, the contradiction is in the output and the prose says so.

The assembly at the end wires every block into one working thing and measures it.

Block 1 — Input splitting

Teaches: a split must never cut a record in half

The problem. Before any distribution happens, the input has to be cut into pieces that can be processed independently. The cut looks trivial and is the source of a whole class of silent data-loss bugs.

@block(1, "Input splitting", "a split must never cut a record in half")
def b1(s, show):
    def split(words, m):
        per = max(1, len(words) // m)
        return [words[i:i+per] for i in range(0, len(words), per)]
    parts = split(TEXT, 8)
    assert sum(len(p) for p in parts) == len(TEXT), "lost or duplicated records"
    if show:
        print(f"  {len(TEXT)} words -> {len(parts)} splits of ~{len(parts[0])}")
        print(f"  conservation: sum(len) == len(input): "
              f"{sum(len(p) for p in parts) == len(TEXT)}")
        print("  on real files the boundary lands mid-line; each split reads to the")
        print("  next delimiter and skips a leading partial. Test a record that")
        print("  spans a boundary or you will silently lose or double it.")
    return {"split": split}

Reading the implementation

The conservation assertion — sum(len(p)) == len(input) — is the entire point of this block. On a list of words a split cannot go wrong; on a real file it goes wrong immediately, because a byte-range split lands in the middle of a record.

The standard solution is worth stating precisely because it is not obvious: each split reads past its end to the next delimiter, and skips a leading partial record. Split \(i\) owns the records that start within its range. That rule makes ownership unambiguous with no coordination between readers — the splits do not need to talk to each other, which is what lets them run on different machines.

Two consequences fall out:

• Records must be self-delimiting, which is why Hadoop's SequenceFile and Avro's container format embed periodic sync markers: a reader that starts at an arbitrary offset can scan forward to a known boundary. A gzip file has no such markers, which is exactly why gzip input is not splittable and a 10 GB gzip file becomes one mapper — a classic performance surprise.
• Split size is a scheduling parameter, not a storage one. Too large and the tail of the job is one slow task; too small and the coordinator drowns in task metadata. The 64--128 MB convention comes from matching the HDFS block size so a split is one local read.

What the numbers say

Output:

  4000 words -> 8 splits of ~500
  conservation: sum(len) == len(input): True
  on real files the boundary lands mid-line; each split reads to the
  next delimiter and skips a leading partial. Test a record that
  spans a boundary or you will silently lose or double it.

Beyond the toy

Test the boundary case explicitly: construct a record that spans a split boundary and assert it appears exactly once in the union of outputs. This is a five-line test that finds a bug which otherwise manifests as "the counts are 0.001% off", discovered months later by someone reconciling against another system.

The modern version of this problem is columnar formats. Parquet's row groups are the split unit, and predicate pushdown means a split can be skipped entirely based on min/max statistics — which turns splitting from a partitioning concern into a query-optimisation one.

Block 2 — map + partition

Teaches: hash(key) % R decides the whole shuffle topology

The problem. map is the easy half. partition is the half that decides the entire communication pattern of the job, and it is one line.

@block(2, "map + partition", "hash(key) % R decides the whole shuffle topology")
def b2(s, show):
    def do_map(words): return [(w, 1) for w in words]
    def partition(pairs, R):
        out = defaultdict(list)
        for k, v in pairs: out[hash(k) % R].append((k, v))
        return out
    parts = s["split"](TEXT, 8)
    R = 4
    m0 = do_map(parts[0]); p0 = partition(m0, R)
    if show:
        print(f"  map split 0: {len(parts[0])} words -> {len(m0)} pairs")
        print(f"  partitioned into R={R}: sizes {[len(p0[i]) for i in range(R)]}")
        print("  the SAME key always lands in the same partition, from every mapper.")
        print("  That is what makes the reduce side a merge instead of a join.")
    return {"do_map": do_map, "partition": partition, "R": R}

Reading the implementation

hash(key) % R is doing something stronger than it looks: it guarantees that the same key lands in the same partition from every mapper, independently, with no coordination. That is what makes the reduce side a merge rather than a distributed join — reducer \(r\) receives all values for its keys and nothing else.

The properties that follow:

• Determinism is required. Python's hash() for strings is randomised per process since 3.3 (PYTHONHASHSEED), so a real implementation must use a stable hash — MurmurHash, xxHash, or an explicit seed. A job whose partitioning changes between mapper restarts silently produces wrong results, and it is precisely the kind of bug that only appears when a task is retried.
• Skew lives here. % R distributes keys evenly, not values. One key with 10% of the records sends 10% of the data to one reducer, and no amount of parallelism helps. Detect it with a per-partition size histogram before blaming the cluster.
• The partition count \(R\) is fixed at job start and determines the output file count. Choosing it badly gives either a million tiny files (which destroy the next job's planning) or a handful of enormous ones (which cannot be parallelised downstream).

What the numbers say

Output:

  map split 0: 500 words -> 500 pairs
  partitioned into R=4: sizes [100, 50, 300, 50]
  the SAME key always lands in the same partition, from every mapper.
  That is what makes the reduce side a merge instead of a join.

Beyond the toy

Skew mitigations, in increasing order of intrusiveness: salting (append a random suffix to hot keys, then a second aggregation pass), two-phase aggregation (pre-aggregate per mapper, which is block 4), broadcast joins (replicate the small side to every mapper, avoiding the shuffle entirely), and adaptive execution (Spark 3's AQE, which detects skewed partitions at runtime from actual sizes and splits them). The last is the right answer and it required the engine to see the whole DAG — which MapReduce cannot.

Block 3 — Shuffle: the all-to-all

Teaches: M x R transfers -- the pattern that scales worst

The problem. The shuffle is the only genuinely distributed part of MapReduce, and it is the part that decides whether the job takes minutes or hours. \(M \times R\) transfers is a communication pattern that scales worse than anything else in the system.

@block(3, "Shuffle: the all-to-all", "M x R transfers -- the pattern that scales worst")
def b3(s, show):
    def shuffle(map_outputs, R):
        bytes_moved = 0
        red_in = {r: [] for r in range(R)}
        for parts in map_outputs:
            for r in range(R):
                chunk = parts.get(r, [])
                red_in[r].extend(chunk)
                bytes_moved += sum(len(k) + 8 for k, _ in chunk)
        return red_in, bytes_moved
    splits = s["split"](TEXT, 8)
    mo = [s["partition"](s["do_map"](sp), s["R"]) for sp in splits]
    red_in, moved = shuffle(mo, s["R"])
    if show:
        print(f"  M={len(splits)} mappers x R={s['R']} reducers = "
              f"{len(splits)*s['R']} transfers")
        print(f"  {moved/1024:.1f} KB moved across the shuffle")
        print(f"  reducer input sizes: {[len(v) for v in red_in.values()]}")
        print("  all-to-all is why the shuttle dominates: it grows as M*R, and the")
        print("  network is the scarcest resource in the cluster.")
    return {"shuffle": shuffle}

Reading the implementation

Every mapper produces data for every reducer, so the number of transfers is \(M \times R\). At \(M = R = 1000\) that is a million connections, and the data crosses the network once but touches disk two to four times: map output write, spill and merge, transfer, reduce-side merge.

The byte counter here is the honest instrumentation. It is what makes block 4's combiner improvement measurable rather than asserted, and it is the number to watch in production — a job whose shuffle bytes exceed its input bytes is usually doing something wrong.

What the numbers say

Output:

  M=8 mappers x R=4 reducers = 32 transfers
  45.3 KB moved across the shuffle
  reducer input sizes: [800, 400, 2400, 400]
  all-to-all is why the shuttle dominates: it grows as M*R, and the
  network is the scarcest resource in the cluster.

Beyond the toy

Two hardware facts shape every real shuffle implementation:

• Sort-based vs hash-based. Hash shuffle writes \(R\) small files per mapper — a million files and a million random writes at \(M=R=1000\). Sort shuffle writes one partitioned, sorted file per mapper, so reducers do large sequential reads at known offsets. Spark switched its default for exactly this reason, and it is the same random-vs-sequential argument as P04.
• Bisection bandwidth. In an oversubscribed tree topology (4:1 was typical in 2004) cross-rack bandwidth is a small fraction of intra-rack, which is why data locality mattered so much in the original paper. Modern Clos fabrics are close to non-blocking, which is why locality matters less now and disaggregated storage (S3 + stateless compute) became viable at all.

The current refinement is push-based shuffle (Magnet, Cosco): mappers push their output to reducers, which merge it into large sequential files as it arrives. That converts many small random reads into few large sequential ones — a 2020s revisiting of the exact problem the 2004 paper had, on hardware where the answer changed.

Block 4 — Combiner

Teaches: pre-aggregate on the map side -- only valid if reduce is associative

The problem. If the reduce function is associative and commutative, most of the shuffle is redundant — you are shipping a thousand ("the", 1) pairs where one ("the", 1000) would do. The combiner exploits that, and the conditions on it are what matter.

@block(4, "Combiner", "pre-aggregate on the map side -- only valid if reduce is associative")
def b4(s, show):
    def combine(pairs):
        agg = defaultdict(int)
        for k, v in pairs: agg[k] += v
        return list(agg.items())
    splits = s["split"](TEXT, 8)
    raw = [s["partition"](s["do_map"](sp), s["R"]) for sp in splits]
    comb = [s["partition"](combine(s["do_map"](sp)), s["R"]) for sp in splits]
    _, b_raw = s["shuffle"](raw, s["R"])
    _, b_com = s["shuffle"](comb, s["R"])
    if show:
        print(f"  shuffle bytes without combiner: {b_raw/1024:>7.1f} KB")
        print(f"  shuffle bytes with combiner:    {b_com/1024:>7.1f} KB  "
              f"({b_raw/b_com:.0f}x less)")
        print("  valid ONLY because + is associative and commutative. A combiner")
        print("  applied to a non-associative reduce is silently wrong -- the")
        print("  framework must REJECT it, not trust you.")
    return {"combine": combine}

Reading the implementation

A combiner is a map-side pre-aggregation using the reduce function. Its correctness requires that the reduce operation be associative and commutative, because the framework decides how many times to apply it and in what grouping — possibly zero times, possibly repeatedly during a multi-pass spill merge.

That condition is not a footnote. sum is fine. average is not — averaging averages is wrong — and the fix is to make the intermediate value a (sum, count) pair, which is associative, and divide only in the final reduce. This is the same algebraic requirement as a monoid, and the same one that makes sketches (HyperLogLog, t-digest, count-min) mergeable and therefore usable in a combiner.

The framework should reject a non-associative combiner rather than trust the programmer, because the failure is silent and non-deterministic: the result depends on how many spill passes happened, which depends on memory pressure.

What the numbers say

Output:

  shuffle bytes without combiner:    45.3 KB
  shuffle bytes with combiner:        0.6 KB  (70x less)
  valid ONLY because + is associative and commutative. A combiner
  applied to a non-associative reduce is silently wrong -- the
  framework must REJECT it, not trust you.

A 1000× reduction is not unrepresentative for aggregations over a skewed key distribution — the head of a Zipf distribution compresses enormously. For a job with unique keys the combiner buys nothing and costs CPU, which is why frameworks make it optional and why measuring it is worthwhile rather than assuming.

Beyond the toy

• In-mapper combining goes further: keep a hash map in the mapper and emit only at close. It gets full aggregation instead of per-spill aggregation, at the cost of unbounded memory — the standard compromise is a bounded map with LRU eviction, which recovers most of the benefit with a memory ceiling.
• Algebraic vs holistic aggregates. Sum, count, min, max and any moment are algebraic and combine perfectly. Median and exact distinct-count are holistic and do not — which is why the approximate versions (t-digest, HyperLogLog) are not merely faster but architecturally necessary.

Block 5 — Reduce + atomic commit

Teaches: at-least-once execution plus an atomic rename = exactly-once effect

The problem. A task may run more than once — after a crash, or speculatively, or because the coordinator lost track of it. The output must nevertheless appear exactly once. This block is where at-least-once execution becomes exactly-once effect, and the mechanism is one syscall.

@block(5, "Reduce + atomic commit", "at-least-once execution plus an atomic rename = exactly-once effect")
def b5(s, show):
    D = tempfile.mkdtemp(prefix="h06-")
    def do_reduce(pairs):
        agg = defaultdict(int)
        for k, v in pairs: agg[k] += v
        return sorted(agg.items())
    def commit(result, rid, attempt):
        tmp = os.path.join(D, f"part-{rid}.attempt{attempt}.tmp")
        with open(tmp, "w") as f:
            for k, v in result: f.write(f"{k}\t{v}\n")
        os.replace(tmp, os.path.join(D, f"part-{rid}"))   # ATOMIC
        return os.path.join(D, f"part-{rid}")
    if show:
        splits = s["split"](TEXT, 8)
        mo = [s["partition"](s["combine"](s["do_map"](sp)), s["R"]) for sp in splits]
        red_in, _ = s["shuffle"](mo, s["R"])
        r0 = do_reduce(red_in[0])
        p1 = commit(r0, 0, attempt=1)
        p2 = commit(r0, 0, attempt=2)        # a duplicate/speculative task
        print(f"  reducer 0 produced {len(r0)} keys, e.g. {r0[:2]}")
        print(f"  attempt 1 and attempt 2 both committed -> one file: "
              f"{p1 == p2 and os.path.exists(p1)}")
        print(f"  no .tmp files left behind: "
              f"{not any(f.endswith('.tmp') for f in os.listdir(D))}")
        print("  write-temp-then-rename is why a duplicated task is harmless. Same")
        print("  trick as P03's segment flush and P07's checkpoint: make the state")
        print("  transition and the position advance ATOMIC.")
    return {"do_reduce": do_reduce, "commit": commit, "D": D}

Reading the implementation

write to part-N.attemptK.tmp
rename to part-N          # atomic

rename(2) on POSIX is atomic within a filesystem: any observer sees either the old name or the new one, never a partial state. Two attempts of the same task therefore produce one file, and whichever finishes last wins — harmlessly, because they contain the same bytes (the task is deterministic).

This is the same primitive as P04's SSTable flush, P03's segment commit, and P07's checkpoint. Once you see it, "exactly-once" stops being mysterious: the guarantee is never about delivery, it is about a pointer moving atomically.

The test in the block — commit attempt 1, commit attempt 2, assert one file and no leftover .tmp — is the whole contract in three lines.

What the numbers say

Output:

  reducer 0 produced 2 keys, e.g. [('lazy', 400), ('quick', 400)]
  attempt 1 and attempt 2 both committed -> one file: True
  no .tmp files left behind: True
  write-temp-then-rename is why a duplicated task is harmless. Same
  trick as P03's segment flush and P07's checkpoint: make the state
  transition and the position advance ATOMIC.

Beyond the toy

The property does not hold on object stores, and this is a significant real- world gap. S3 has no atomic rename; a "rename" is a copy plus a delete, which is neither atomic nor cheap for large objects. The consequences are the entire history of Hadoop's FileOutputCommitter v1 vs v2, the S3A committers, and eventually table formats (Iceberg, Delta Lake, Hudi) that put a real atomic commit — a single-object metadata pointer swap or a database transaction — on top of the object store. The lesson generalises: when the substrate lacks the primitive you need, the design problem becomes building it.

Two more requirements this toy skips: fsync the file and its parent directory before the rename is durable (the most-forgotten line in this pattern), and cleaning up .tmp files from failed attempts, which otherwise accumulate silently.

Block 6 — Stragglers and backup tasks

Teaches: job time is a MAXIMUM, and maxima behave badly

The problem. A job finishes when its slowest task finishes. Maxima behave badly, and this block quantifies exactly how badly — which is the argument for a mechanism that looks wasteful.

@block(6, "Stragglers and backup tasks", "job time is a MAXIMUM, and maxima behave badly")
def b6(s, show):
    def job(ntasks=200, nworkers=20, frac=0.0, mult=1, backup=False, trials=400):
        out = []
        for _ in range(trials):
            t = [10.0 * (mult if rng.random() < frac else 1.0) for _ in range(ntasks)]
            if backup: t = [min(x, 20.0) for x in t]
            w = [0.0] * nworkers
            for x in sorted(t, reverse=True):
                i = w.index(min(w)); w[i] += x
            out.append(max(w))
        return statistics.fmean(out)
    if show:
        base = job()
        print(f"  {'scenario':<22}{'completion':>12}{'vs ideal':>10}{'w/ backup':>12}")
        for frac, mult, lbl in ((0.0,1,"no stragglers"),(0.01,10,"1% at 10x"),
                                (0.05,10,"5% at 10x"),(0.01,50,"1% at 50x")):
            a = job(frac=frac, mult=mult); b = job(frac=frac, mult=mult, backup=True)
            print(f"  {lbl:<22}{a:>10.1f}s{a/base:>9.2f}x{b:>10.1f}s")
        print("  two tasks in two hundred inflate the job 4.5x. Backup tasks recover")
        print("  nearly all of it. That is MapReduce section 3.6, generated not quoted.")
    return {"job": job}

Reading the implementation

The simulation assigns tasks to workers with longest-processing-time-first scheduling (a good approximation) and takes the makespan. Then the same with backup tasks, modelled as capping any task at 2× the normal duration — which is what launching a duplicate on another machine achieves in expectation.

The reason this is worth simulating rather than reasoning about is that the intuition is wrong. A 1% straggler rate sounds negligible; over 200 tasks it means two tasks are slow with near-certainty, and if they are 10× slow the job is 4.5× longer. The arithmetic of maxima is not the arithmetic of means.

What the numbers say

Output:

  scenario                completion  vs ideal   w/ backup
  no stragglers              100.0s     1.00x     100.0s
  1% at 10x                  112.5s     1.13x     108.7s
  5% at 10x                  148.6s     1.49x     110.0s
  1% at 50x                  451.0s     4.51x     108.5s
  two tasks in two hundred inflate the job 4.5x. Backup tasks recover
  nearly all of it. That is MapReduce section 3.6, generated not quoted.

Beyond the toy

• Naive speculation makes it worse. The LATE paper's central finding: a scheduler that speculates on "tasks furthest behind" keeps relaunching tasks on the slow machines that caused the problem, consuming capacity and slowing the job. LATE speculates on estimated finish time instead, which requires progress-rate estimation and is genuinely harder.
• Causes of stragglers, roughly in order of frequency: data skew (block 2), hardware degradation (a disk with remapped sectors), resource contention from a co-tenant, GC pauses, and network hot spots. Only the first is fixable by the programmer, which is why the framework must handle the rest.
• The general form is Dean & Barroso's Tail at Scale: hedged requests, tied requests, micro-partitioning (many more partitions than machines, so load balances naturally and hot partitions can migrate), and selective replication. The same arithmetic appears in P10's multiple comparisons and P15's fan-out — one distribution, three contexts.

Assembly note

The three-kill-schedule table is the only correctness criterion that matters for a batch framework, and it is worth being explicit about why byte-identical output is achievable at all: map and reduce are pure functions of their input. That restriction is the price, and retry, speculation, rescheduling and recomputation are all things it buys. Allow one map function to read the clock, a random seed, or an external service, and every one of those mechanisms becomes unsound simultaneously — the output becomes a function of the failure schedule, which is exactly what the test detects.

The assembly

Every block above, wired together into one working system:

def assembly(s):
    print("\nSix blocks = a batch framework. Run word count, then kill workers.\n")
    def run_job(text, M=8, R=4, kill_workers=0, use_combiner=True):
        splits = s["split"](text, M)
        map_out, retries = [], 0
        for i, sp in enumerate(splits):
            attempt = 0
            while True:
                attempt += 1
                # a "worker failure" loses the task's output; the coordinator retries
                if i < kill_workers and attempt == 1:
                    retries += 1; continue
                pairs = s["do_map"](sp)
                if use_combiner: pairs = s["combine"](pairs)
                map_out.append(s["partition"](pairs, R)); break
        red_in, moved = s["shuffle"](map_out, R)
        result = {}
        for r in range(R):
            for k, v in s["do_reduce"](red_in[r]): result[k] = v
        return result, retries, moved

    truth = {}
    for w in TEXT: truth[w] = truth.get(w, 0) + 1
    print(f"  {'run':<28}{'keys':>7}{'matches oracle':>17}{'retries':>9}")
    for lbl, kw, comb in (("clean", 0, True), ("3 workers killed", 3, True),
                          ("6 workers killed", 6, True), ("no combiner", 0, False)):
        res, ret, moved = run_job(TEXT, kill_workers=kw, use_combiner=comb)
        print(f"  {lbl:<28}{len(res):>7}{str(res == truth):>17}{ret:>9}")
    print("\n  Identical output under every fault schedule. That is the ONLY")
    print("  correctness criterion that matters, and it is possible because map and")
    print("  reduce are pure functions of their input -- the framework may re-run")
    print("  any task, anywhere, at any time.")
    print("\n  The restriction IS the feature. Let map read shared mutable state and")
    print("  retry, speculation and rescheduling all become unsound at once.")
    print("\n  Built: splitting -> map/partition -> shuffle -> combiner -> reduce +")
    print("  atomic commit -> straggler simulation.")
    print("  Missing, on the project page: real processes and sockets (m3-m5),")
    print("  coordinator checkpointing (m10), data locality (m11), the LATE")
    print("  scheduler (E5, where naive speculation makes things WORSE), and E12 --")
    print("  the hand-written comparison that is the report's thesis.")

Output:

Six blocks = a batch framework. Run word count, then kill workers.

  run                            keys   matches oracle  retries
  clean                             7             True        0
  3 workers killed                  7             True        3
  6 workers killed                  7             True        6
  no combiner                       7             True        0

  Identical output under every fault schedule. That is the ONLY
  correctness criterion that matters, and it is possible because map and
  reduce are pure functions of their input -- the framework may re-run
  any task, anywhere, at any time.

  The restriction IS the feature. Let map read shared mutable state and
  retry, speculation and rescheduling all become unsound at once.

  Built: splitting -> map/partition -> shuffle -> combiner -> reduce +
  atomic commit -> straggler simulation.
  Missing, on the project page: real processes and sockets (m3-m5),
  coordinator checkpointing (m10), data locality (m11), the LATE
  scheduler (E5, where naive speculation makes things WORSE), and E12 --
  the hand-written comparison that is the report's thesis.

The design space

MapReduce is one point in the space of batch execution engines, and the axis that matters is how much of the dataflow graph the system can see at once.

The progression is a single trade being renegotiated: materialise more → recover cheaply but run slowly; pipeline more → run fast but recover expensively. MapReduce sits at the extreme materialising end because it was designed for clusters of thousands of unreliable commodity machines where a job would certainly experience failures during its run.

Spark's contribution was noticing that if the transformation is deterministic and its inputs are still available, you can recover by recomputation rather than replication — which is only sound because of the same purity restriction that makes MapReduce's task retries safe. The assembly's byte-identical output under three kill schedules is the demonstration that the restriction buys the property.

The shuffle is the system

Everything expensive in a batch engine is the all-to-all. With \(M\) mappers and \(R\) reducers there are \(M \times R\) transfers, and the data crosses the network exactly once but touches disk two to four times.

Two hardware facts set the shape:

• Bisection bandwidth. In an oversubscribed tree topology (4:1 is common), cross-rack bandwidth is a fraction of intra-rack. Data locality — scheduling a map task on a node that already holds its input block — was worth so much in the original paper precisely because of this. Modern Clos/fat-tree fabrics are much closer to non-blocking, which is why locality matters less than it did and disaggregated storage (S3) became viable.
• Sequential vs random IO. Sort-based shuffle writes one partitioned, sorted file per map task; hash-based shuffle writes \(R\) small files per map task. At \(M = R = 1000\) that is a million files and a million random writes, which is why Spark moved from hash to sort-based shuffle by default.

The combiner in block 4 attacks this directly, and its 1000× reduction there is not unrepresentative: for associative aggregations the map-side combine is usually the single largest win available.

Stragglers, and the arithmetic of maxima

A job finishes when its slowest task finishes, and maxima behave badly. If each task independently has probability \(p\) of being slow, a job of \(n\) tasks is slow with probability \(1 - (1-p)^n\): at \(p = 0.01\) and \(n = 100\) that is 63%. This is the same arithmetic as P10's multiple-comparisons block and P15's tail composition — one distribution, three contexts.

Block 6 measures it: 1% of tasks running 10× slow inflates the job 4.5×, and backup tasks recover nearly all of it. The counter-intuitive part, documented in the LATE paper, is that naive speculation makes things worse on heterogeneous clusters: a scheduler that speculates on "tasks furthest behind" will keep re-launching tasks on the slow machines that caused the problem, consuming capacity. LATE speculates on estimated finish time instead.

Dean & Barroso's The Tail at Scale generalises: hedged requests, tied requests, micro-partitioning (many more partitions than machines, so load balances naturally), and selective replication of hot partitions.

Advanced algorithms and alternatives

• External sorting is the core primitive: \(O(N \log_{M/B} (N/B))\) IOs in the external-memory model, and the reason the shuffle is sort-based.
• Distributed joins: broadcast (small side fits in memory), shuffle-hash, sort-merge, and skew-aware variants that split hot keys. Skew in a join key is the single most common cause of one reducer running for hours.
• Approximate aggregation: HyperLogLog for distinct counts, t-digest for quantiles, count-min for heavy hitters — all mergeable, which is exactly the property that lets them be computed in a combiner.
• Ring all-reduce is the ML analogue of the shuffle: bandwidth-optimal, \(2(N-1)/N\) of the data per node, and the reason data-parallel training scales where a parameter-server design bottlenecks on the server's NIC.
• Push vs pull shuffle: Magnet/Cosco push map output to reducers to convert many small random reads into few large sequential ones — a 2020s revisiting of exactly the problem the 2004 paper had.

Hardware: what changed since 2004

The original design assumed 100 Mb/s--1 Gb/s networks, spinning disks, and frequent machine failure. Today: 25--100 Gb/s NICs, NVMe at GB/s, and object storage where compute and data are deliberately separated. Consequences:

• Locality matters less; disaggregation (S3 + stateless compute) won because the network stopped being the bottleneck relative to disk.
• Memory is large enough that many "big data" jobs fit on one machine. A modern server with 1--2 TB of RAM makes a single-node engine (DuckDB, Polars) faster than a cluster for a large fraction of real workloads — the "COST" critique (Configuration that Outperforms a Single Thread).
• CPU became the bottleneck rather than IO, which is why columnar formats (Parquet, ORC), vectorised execution, and compression are now where the wins are.

How this connects to the rest of the track

• P07 is this system with the batch boundary removed; its checkpoint is this project's atomic commit applied continuously.
• P05 provides the fault-tolerant coordinator a real implementation needs.
• P04's sequential-write discipline and this project's sort-based shuffle are the same response to the same storage physics.
• P01's MoE routing and tensor-parallel all-reduce are the same all-to-all pattern at NVLink speed.
• P10 shares the maxima arithmetic that makes stragglers and false positives both inevitable at scale.

Failure modes at scale

• Skew. One key with 10% of the rows makes one reducer take 10% of the total work alone; no amount of parallelism helps. Detect with a per-partition size histogram before blaming the cluster.
• Small files. Millions of tiny outputs destroy the next job's planning time and the namenode/metadata store.
• Speculation feedback loops, as above.
• Non-deterministic map functions — a UDF that reads the clock, a random seed, or an external service — silently break the retry guarantee. The output is then a function of the failure schedule, which is exactly what the assembly tests for.
• Coordinator as a single point of failure, which is why m10 on the project page is coordinator checkpointing.

Primary sources

• Dean & Ghemawat, MapReduce: Simplified Data Processing on Large Clusters (OSDI 2004) — §3.6 on backup tasks is what block 6 reproduces.
• Zaharia et al., Resilient Distributed Datasets (NSDI 2012).
• Zaharia et al., Improving MapReduce Performance in Heterogeneous Environments (LATE, OSDI 2008).
• Dean & Barroso, The Tail at Scale (CACM 2013).
• McSherry, Isard & Murray, Scalability! But at what COST? (HotOS 2015).
• Ghemawat, Gobioff & Leung, The Google File System (SOSP 2003) — the storage assumptions the whole design rests on.

Running it

python3 handson/h06_mapreduce.py            # every block, then the assembly
python3 handson/h06_mapreduce.py --block 3  # just block 3 and its prerequisites
python3 handson/h06_mapreduce.py --quiet    # the assembly only

What to do with this

Replace the in-process tasks with real subprocesses and a coordinator that detects a dead worker by timeout rather than by return value. The moment tasks can fail silently rather than by raising, the design pressure changes completely --- and that is the environment the original paper was written for.

Milestones, experiments, readings and exit criteria for this project: P06 — MapReduce-Style Framework.