Batch: Your Own MapReduce
Immutable inputs, deterministic tasks, and an atomic rename turn failure from a protocol problem into a non-event.
The idea
Everything so far has been about services: requests arrive, you answer fast, state mutates in place, and failure handling is agony. Batch processing flips all of that. A job reads a pile of read-only input files, grinds for a while, and writes a fresh pile of output files, never touching its input. That single rule makes a job a pure function of its data, so you can re-run it, debug it against yesterday's files, or throw a bad output away and try again.
MapReduce is just the Unix pipeline taught to run on many machines. Map runs a function over every input record independently and emits key-value pairs. Shuffle routes each pair to a partition by hashing its key and sorts each partition. Reduce walks the sorted partition and handles each key's now-adjacent values together. Map tasks never talk to each other and neither do reduce tasks, so all the coordination in the model is concentrated in the shuffle.
The shuffle is a meeting point, which is what makes joins possible without a database. Every record with the same key lands in the same place, already sorted, so grouping is free and a join becomes two mapped inputs zipped together by one reducer. Sorting and merging files replaces indexes and random I/O entirely. Underneath it all sits external merge sort: sort what fits in memory, spill it to disk, then merge the runs.
Fault tolerance falls out almost for free, and it is earned rather than lucky. Inputs are immutable so a dead task destroyed nothing, tasks are deterministic so a retry produces the same bytes, and output is written to a temp file and atomically renamed only on success. The cracks show up elsewhere: one hot key can leave a single reducer grinding while its peers idle, and writing every intermediate result to disk is why dataflow engines like Spark exist.
The bench — 4 exercises
The Teaspoon and the Mountain
Sort a file far larger than your memory budget by building external merge sort yourself, the primitive that every later piece of this module runs on.
- Generate a multi-gigabyte web-server log and answer 'top 5 most requested URLs' with a Unix one-liner, noting the wall time
- Write a sorter that reads input in chunks sized to fit a memory budget, sorts each chunk in memory, and spills it to a numbered temp run file
- Merge all run files with a k-way min-heap over one buffered head record per run, writing a single sorted output
- Run it under a hard memory cap (ulimit -v or a cgroup) that is a small fraction of the input size
- Expose the sorter as both a library function and a standalone binary, since the next exercise will call it
hint
The classic k-way merge bug is comparator direction: min-heap by record key, tie-broken by run index for stability.
hint
Give every run file its own read buffer. Unbuffered per-record reads make the merge phase dwarf the sort phase.
DONE WHEN
· Sort completes without ever exceeding the memory cap
· Output is byte-identical to the system sort's output on the same file
· At least eight run files exist in the temp directory at peak, proving spilling really happened
· Both the one-liner and your program report the same top 5 URLs
Build the Framework
Assemble a coordinator, workers, and a hash-partitioned shuffle into a real mini-MapReduce, then certify it by building an inverted index you can actually query.
- Write a coordinator that splits input into roughly 64 MB chunks and hands tasks to workers polling over a local socket
- Implement map tasks: run the user function over one split and hash-partition emitted pairs into R sorted intermediate files named mr-<map>-<part>
- Implement reduce tasks: fetch that partition from every map output, k-way merge them with your sorter, group keys, and write out-<part>
- Register jobs behind a --job=<name> flag compiled into the worker rather than dynamically loading plugins
- Certify with word count first, then an inverted index over a document corpus, and write a tiny CLI that answers term queries from the index files
hint
Keep the coordinator dumb and tasks pure: everything a task needs must be in its task spec, so any worker can run any task cold.
hint
If reduce output is wrong only for keys near a buffer boundary, your key-group iterator is comparing prefixes instead of full keys.
DONE WHEN
· Concatenated and sorted output matches a single-process reference implementation
· Task assignment timestamps show at least three workers busy at once during the map phase
· Every key appears in exactly one partition, matching hash(key) % R
· The query CLI returns correct posting lists for three sample terms
Joins and the Celebrity Problem
Run a sort-merge join and a top-K query on the framework unchanged, watch one hot key create a straggler, then fix it with two-stage aggregation.
- Generate a users file and a much larger activity file with Zipf-distributed user IDs
- Map both inputs to (user_id, tagged record), shuffle together, and reduce into per-user activity summaries including users with zero events
- Log per-task wall times and find the reduce task taking many times the median: that is the celebrity's partition
- Implement two-stage aggregation for top 5 URLs per user-agent, salting hot keys in stage one and stripping salts in stage two
- Implement a broadcast join variant that loads the small side into every mapper's memory, and compare it against the sort-merge join under different memory caps
hint
Sort profiles before events within a key by putting a tag in the sort key but stripping it before partitioning, so the reducer holds one profile and streams events in constant memory.
hint
Keep the naive single-stage path behind a flag. The straggler measurement is part of the lesson, not a bug to delete.
DONE WHEN
· Join output includes users with zero events and matches a reference join
· Naive plan shows a max reduce-task time at least four times the median
· Two-stage output is identical to the naive answer with max task time under twice the median
· Broadcast join output matches sort-merge byte for byte when memory allows, and fails cleanly without partial output when it does not
Kill the Workers
Make the join job survive random kill -9s and still produce byte-identical output, proving that determinism plus atomic rename replaces any failure protocol.
- Track per-task state (idle, in-progress, done) in the coordinator with a liveness timeout that reassigns silent workers' tasks
- Have every task write to tmp-<task>-<attempt> and atomically rename to an attempt-independent final name only on success
- Deduplicate by task ID so a task completed twice simply renames the same bytes over itself
- Run the job repeatedly while killing random workers at different phases, including after a worker finishes but before it reports
- Compare each chaotic run's output against a clean run of the same input
hint
Final names must not contain the attempt number, or a second attempt lands beside the first instead of over it.
hint
Too short a liveness timeout re-runs slow-but-alive tasks (harmless thanks to the rename), too long stalls the job on one dead worker. Measure rather than guess.
DONE WHEN
· Jobs complete despite at least three worker kills per run
· Output checksum matches the clean run every time
· No tmp- files remain and no final output file is duplicated
· A run that kills every worker at least once still finishes
Go deeper (after the bench)
Read DDIA ch. 10 now, especially the Unix philosophy opening and the reduce-side versus map-side join treatment, which is exercise 03 with production war stories attached. Then read the original MapReduce paper by Dean and Ghemawat (Google, OSDI 2004, free PDF) — thirteen pages that will read like a design review of the code you just wrote, including its re-execution semantics and the combiner trick you rediscovered as two-stage aggregation.