P06 — MapReduce-Style Computation Framework

Run it first. There is a companion page that builds this project's machinery as numbered, independently runnable blocks and then assembles them into one measured system: P06 hands-on — block by block (handson/h*.py). Every number on it was produced by running the code. Read it alongside the milestones below.

Medium · 88 hours · Weeks 68–75 · Stage 3 · Go

Table of Contents

• The Loop, Instantiated
• Why This Project Matters
• Prerequisites
• Duration and Size
• Central Technical Questions
• Architecture
• Showcase — Why Backup Tasks Exist
• Implementation Milestones
• Concepts To Study
• Primary-Source Readings
• Experiments
• Benchmarks and Metrics
• Correctness Tests
• Failure Tests
• Expected Difficulties
• Scope Boundaries
• Deliverables
• Exit Criteria
• Extension Ideas
• Connections
• References

The Loop, Instantiated

Why This Project Matters

MapReduce's contribution was not the algorithm — grouping by key and aggregating is older than computing. It was the observation that if you restrict what a programmer may express, you can automate fault tolerance.

Because a map function is required to be a pure function of one input record, and a reduce function a pure function of one key's values, the framework knows that any task can be re-executed anywhere at any time with the same result. That single property is what lets it retry, speculate, and reschedule without the user writing a line of recovery code. Take the restriction away — let map read shared mutable state — and every one of those mechanisms becomes unsound.

This is the most transferable idea in the whole journey. The same trade appears in React's pure render functions, in Spark's lineage-based recovery, in Terraform's declarative resources, in every idempotent HTTP API you have designed, and in functional programming generally. Restriction buys automation. This project is where you feel the price and the payoff at the same time, because you will write both the framework and the hand-rolled distributed alternative and compare them.

Prerequisites

• P05 complete — the fault injector, membership, and failure detector are reused directly
• Go concurrency; comfort with process-level parallelism
• Familiarity with EMR/Hive from your day job is an asset here: you have operated this model for years and can now build it

Duration and Size

Medium, 88 hours, 8 weeks.

Central Technical Questions

1. What must be true of a map function for the framework to retry it freely? State the contract precisely.
2. Why is the shuffle the hard part? It is an all-to-all data movement, and all-to-all is the pattern that scales worst.
3. What is the right task size? Too large and stragglers dominate; too small and coordinator overhead does. Derive both bounds.
4. Why does one hot key ruin everything, and what are the four available fixes?
5. How much does speculative execution actually buy, and when does it cost more than it saves?
6. What does the coordinator's failure cost you, and what would it take to survive it?

Architecture

Write your naive design first.

   input files ──► split into M chunks (64 MB nominal)
                          │
        ┌─────────────────┴──────────────────┐
        ▼                                    ▼
   ┌─────────┐                          ┌─────────┐
   │ map     │  emit(k,v) ──► partition │ map     │   R local files each
   │ task 1  │  by hash(k)%R ──► sort ──│ task M  │   (the shuffle write)
   └────┬────┘                          └────┬────┘
        └────────────┬───────────────────────┘
                     │  reduce task r pulls its partition from ALL M mappers
                     ▼          (this is the all-to-all: M × R transfers)
              ┌──────────────┐
              │ reduce task r│  merge-sort R inputs ──► reduce(k, [v]) ──► output
              └──────────────┘

   coordinator: task table (idle/in-progress/done), worker heartbeats,
                retry on failure, speculative duplicate on slow, atomic rename on commit

Task granularity, derived

Let \(M\) be the number of map tasks, \(W\) workers, \(c\) the per-task coordinator overhead, and \(T\) the total work.

• Per-task overhead costs \(Mc\) in total. Larger \(M\) → more overhead.
• Straggler exposure: job time is bounded below by the slowest single task, roughly \(T/M\) for balanced work. Larger \(M\) → smaller tail exposure.
• Load balance: with \(M \gg W\), dynamic assignment smooths worker speed differences. The classic rule of thumb is \(M \approx 10W\) or more.

So \(M\) is squeezed from both sides, and the optimum depends on \(c\), which you must measure. The original paper used M=200,000, R=5,000 on 2,000 machines — that is 100 map tasks per machine, and the ratio is the interesting part, not the absolute numbers.

Why stragglers dominate, quantified

Job completion is the maximum over tasks, not the mean, and maxima behave badly. Simulated: 200 tasks of 10 s nominal on 20 workers, greedy list scheduling, 2,000 trials:

Two tasks out of two hundred, at 50× speed, inflate the job 4.5×. Backup tasks recover almost all of it. That is why §3.6 of the paper exists, and it is a far more persuasive argument as a table you generated than as a sentence you read.

The same effect at the request level is Dean & Barroso's tail-at-scale result. If a request touches \(N\) independent components, each exceeding its p99 latency 1% of the time, the probability that at least one is slow is \(1 - 0.99^N\):

At 100 components, the majority of requests hit a p99 event. Tail latency is not an edge case at scale; it is the common case. Memorise this table.

Skew, and the four fixes

Hash partitioning assumes keys are roughly uniform. Real key distributions are Zipfian: in a news corpus, "the" may be 5% of all tokens, and every occurrence lands on one reducer. Options:

1. Combiners — pre-aggregate on the map side. Works only for associative and commutative reduce functions. Cheapest and most effective when applicable.
2. Salting — append a random suffix to hot keys, reduce in two passes. Costs a second shuffle.
3. Range partitioning with sampling — sample the keyspace, choose boundaries to equalise bytes rather than key count. What TeraSort does.
4. Skew-aware splitting — detect hot keys at runtime and give them dedicated reducers.

Implement 1 and at least one other, and measure both against the unmitigated case.

Showcase — Why Backup Tasks Exist

Forty minutes, before you build a coordinator. Job completion is the maximum over tasks, and maxima behave badly. Simulate it and the case for §3.6 makes itself.

# P06 -- why one straggler in two hundred tasks costs 4.5x.
import random, statistics
random.seed(0)
def job(ntasks=200, nworkers=20, frac=0.0, mult=1, backup=False, trials=800):
    out=[]
    for _ in range(trials):
        t=[10.0*(mult if random.random()<frac else 1.0) for _ in range(ntasks)]
        if backup: t=[min(x, 20.0) for x in t]      # duplicate: cap at 2x nominal
        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)
base=job()
print(f"{'scenario':<28}{'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:<28}{a:>10.1f}s{a/base:>9.2f}x{b:>10.1f}s")
print("\\nTwo tasks out of two hundred inflate the job 4.5x. That is MapReduce 3.6.")

scenario                      completion  vs ideal   w/ backup
no stragglers                    100.0s     1.00x     100.0s
1% at 10x                        112.4s     1.12x     108.5s
5% at 10x                        149.5s     1.50x     110.0s
1% at 50x                        447.5s     4.47x     108.5s
\nTwo tasks out of two hundred inflate the job 4.5x. That is MapReduce 3.6.

Two tasks out of two hundred inflate the job 4.5×, and backup tasks recover almost all of it. That table is a far more persuasive argument for speculative execution than the paragraph in the paper, because you generated it. It is also E4 before you have written a line of the framework.

Implementation Milestones

Concepts To Study

• The map/reduce contract: purity, determinism, and what each buys the framework
• Input splitting and record boundaries — the unglamorous detail that breaks correctness silently
• The shuffle: sort-based vs hash-based; why sorting on the map side makes the reduce side a merge
• External merge sort: k-way merge, the memory/pass trade
• Partitioning: hash, range, custom; sampling for range boundaries
• Combiners and the associativity requirement
• Stragglers and backup tasks; progress-rate estimation
• Task granularity and the derivation above
• At-least-once execution + idempotent commit = exactly-once effect. The atomic rename is what makes this work, and it is the same trick as P03's flush
• Coordinator fault tolerance: checkpointing vs re-execution
• Data locality and the memory/network hierarchy at cluster scale
• Lineage (Spark) as an alternative to re-execution from input

Primary-Source Readings

Budget: 11 hours.

Experiments

E12 is the point of the project. Write word count as a hand-rolled distributed Go program with manual retry. Then write it as a map and a reduce function. Compare: lines of code, what happens when you kill a worker, and how long each took to make correct. The framework version will be slower and dramatically more robust, and articulating why that trade is usually right is the report's thesis.

E5 is the counterintuitive one. Naive speculation ("duplicate any task below the average progress rate") assumes homogeneous workers. On a heterogeneous cluster it duplicates every task on the slow machines, consuming the capacity that would have finished them. It makes things worse, and measuring that is a genuinely good result.

Benchmarks and Metrics

Correctness Tests

1. Output equals single-process output, byte-for-byte after sorting, for every job in the benchmark suite.
2. Under every fault schedule. Same assertion, with the P05 injector running.
3. No duplicate output despite at-least-once task execution — the atomic-rename property.
4. Record boundaries: a record spanning a split boundary appears exactly once.
5. Partition completeness: every emitted key lands in exactly one reduce partition.
6. Sort order within each reduce input.
7. Combiner equivalence: results identical with and without the combiner. A non-associative reduce fn must be rejected, not silently mis-computed.
8. Deterministic replay: same seed, same fault schedule, same output.
9. Empty inputs, single-record inputs, one-key inputs all work.

Failure Tests

The partitioned-worker case is the important one: a worker that loses contact, continues working, and then reappears must not overwrite the output of the replacement task. The atomic rename plus a task-attempt id in the filename is the standard fix, and you should construct the interleaving deliberately to prove yours works.

Expected Difficulties

1. The shuffle is 60% of the work. Budget accordingly; milestones 3–5 are the core.
2. "Distributed" on one laptop needs discipline. Use real processes and real sockets, throttle bandwidth deliberately (E9), and never let a shortcut assume shared memory.
3. Speculative execution can make things worse and you will see it (E5). That is the result, not a bug.
4. Exactly-once output is subtler than it looks. The atomic rename must be paired with attempt ids, or a resurrected worker clobbers good output.
5. Skew experiments need realistic data. Synthetic uniform keys teach nothing; use real text or a Zipfian generator with a stated α.
6. The coordinator is a single point of failure and making it not one is a whole subproject. Milestone 10 is checkpoint-and-restart, not a replicated coordinator. Say so in the report.

Scope Boundaries

In scope: batch jobs, map/shuffle/reduce, task-level fault tolerance, speculative execution, coordinator checkpointing, local-filesystem intermediate storage.

Out of scope: a distributed filesystem (use local disks and simulate locality); a general DAG engine (extension); SQL or a query optimiser; resource negotiation à la YARN; streaming (P07); iterative-job optimisation; a web UI beyond a status endpoint.

Deliverables

1. mapreduce/ — Go framework with the five benchmark jobs
2. REPORT.md whose thesis is E12: why the restricted model wins
3. The worker-utilisation timeline plot — the most legible artifact this project produces, and the one that makes stragglers obvious to any audience
4. Notebook entries for E4, E5, E12
5. Reusable: the straggler simulator, and the skew-aware partitioner

Exit Criteria

• All five benchmark jobs produce output identical to single-process
• Job completes correctly with 50% of workers killed mid-run
• Coordinator failure recovery works from checkpoint
• E4 complete: speculative execution measured across the straggler matrix, compared against the simulated prediction
• E5 complete: naive speculation shown to hurt on a heterogeneous cluster, quantified
• E3 + E11 complete: skew measured and at least two mitigations compared
• E12 complete: the hand-written comparison, with lines of code and fault behaviour
• The partitioned-worker output-clobbering interleaving is tested explicitly
• REPORT.md written with a falsified prediction

Extension Ideas

• Lineage-based recovery (RDD-style): recompute only the lost partition instead of re-running from input. Measure recovery cost on an iterative job — that is the argument of the Spark paper and you can reproduce it.
• General DAG execution with pipelined stages.
• Adaptive task sizing: split slow tasks at runtime rather than duplicating them. A research direction.
• In-memory shuffle with spill-to-disk, and the resulting memory/speed frontier.

Connections

Backward: P05 supplies the fault injector, heartbeats, and failure detection. P04 supplies intermediate storage patterns and the sorted-file merge.

Forward:

• → P07 (Streaming): the scheduler and worker pool generalise to long-running operators; the shuffle becomes a network partitioner
• → P08/P09: batch embedding generation and offline evaluation are natural MapReduce jobs, and running them on your own framework is a good integration check
• → P15: the batch layer

References

• Dean, J., Ghemawat, S. MapReduce: Simplified Data Processing on Large Clusters. OSDI 2004.
• Ghemawat, S., Gobioff, H., Leung, S.-T. The Google File System. SOSP 2003.
• Zaharia, M. et al. Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing. NSDI 2012.
• Zaharia, M., Konwinski, A., Joseph, A. D., Katz, R., Stoica, I. Improving MapReduce Performance in Heterogeneous Environments. OSDI 2008.
• Dean, J., Barroso, L. A. The Tail at Scale. CACM 56(2), 2013.
• Isard, M., Budiu, M., Yu, Y., Birrell, A., Fetterly, D. Dryad: Distributed Data-Parallel Programs from Sequential Building Blocks. EuroSys 2007.
• Verma, A. et al. Large-scale cluster management at Google with Borg. EuroSys 2015.
• Vavilapalli, V. K. et al. Apache Hadoop YARN: Yet Another Resource Negotiator. SoCC 2013.
• Kwon, Y. et al. SkewTune: Mitigating Skew in MapReduce Applications. SIGMOD 2012.
• O'Malley, O. TeraByte Sort on Apache Hadoop. 2008. Range partitioning by sampling.