Part 8 · Derived history

Batch Processing: MapReduce, Joins, Materialization, and Recompute

The failure part (lesson 17) closed the question of keeping a single mutable dataset correct under partial failure, unreliable clocks, consensus, and verification — the source of truth, defended. But almost nothing a product shows you is the raw source of truth. A search index, a recommendation table, a feature store, a daily metrics dashboard, a model's training set — these are all derived from the source data by a computation, and they go stale the moment the source changes. This part is about derived history: recompute a whole view from the log of facts, deterministically, whenever the source moves. This first lesson takes the simplest case: the input is bounded — a finite snapshot you can scan end to end — and you transform all of it at once into a new output. That is batch processing, and its defining superpower is that you can throw the output away and recompute it from the input deterministically.

1 · The model: bounded input, deterministic output

Batch processing is a computation over a bounded input: a finite dataset that already exists in full when the job starts — yesterday's event log, a snapshot of a table, a directory of files. The job reads all of it, transforms it, and writes a new output dataset. The defining contrast is with stream processing (lesson 19), where the input is unbounded — events keep arriving forever and the job never finishes. Bounded means the job has a clear start and a clear end, and crucially, the input does not change while the job runs: it is an immutable snapshot. That immutability is what makes everything else possible.

The intellectual ancestor is the Unix pipeline. A command like cat log | grep ERROR | sort | uniq -c | sort -rn is already a batch job: each tool reads input, does one transformation, and writes output. They compose freely because every tool speaks the same uniform interface — lines of text on stdin/stdout — and none of them mutates its input. Re-run the whole pipeline on the same file and you get the same answer.

MapReduce takes that exact philosophy and scales it across thousands of machines. The uniform interface becomes key-value records in files, and the computation is split into three phases:

Map is the narrow, local, parallel part; reduce is the per-key aggregation; the shuffle in the middle is the all-to-all data movement that connects them. Everything hard and expensive about batch processing lives in that middle step.

Where the input lives: a distributed filesystem and data locality

MapReduce reads its bounded input from a distributed filesystem — the canonical one is HDFS (the Hadoop Distributed File System; cloud object stores like S3 play the same role today). The idea is to spread one logical file across many machines' local disks while presenting it as a single file. A large file is split into fixed-size blocks (commonly 128 MB), and each block is replicated onto several nodes (typically 3) so the data survives a disk or machine failure — the same redundancy argument as replication in Part II, applied to a filesystem.

That layout enables the principle that makes MapReduce scale: data locality, or “bring the compute to the data.” Rather than ship gigabytes of input across the network to wherever a worker happens to be free, the scheduler does the reverse — it places each map task on a machine that already holds a replica of that block, so the map reads from a local disk and nothing crosses the network. Network bandwidth is the scarce resource in a cluster; moving the (small) code to the (large) data instead of the data to the code is what keeps the read phase cheap. Only the shuffle, which genuinely must move data, pays the network cost — which is exactly why the next section is about shrinking it.

2 · The shuffle is a repartition, and it is the tax

You have already met the shuffle under another name. In lesson 11 we partitioned a dataset by a key so that each node owns a slice; reads and writes for a key route to the node that owns that key's partition. The shuffle is exactly that operation, performed on the fly: the mappers' output is repartitioned by the reduce key, so all records for a given key end up co-located on one reducer. Map-side filtering and projection (dropping rows and columns you don't need) are "narrow" — each output partition depends on one input partition. The shuffle is "wide" — every reducer pulls from every mapper. That all-to-all pattern is why it dominates the cost.

This is why every batch optimization you have ever heard of is really "shrink the shuffle": filter early (drop rows before they cross the network), project early (carry only the columns you need), pre-aggregate on the map side (a "combiner" that sums partial counts locally so each key ships one value instead of thousands), and pick the join strategy that avoids the shuffle entirely when you can — which is the next section. When a Spark job is slow, the diagnostic question is always: is it scan-bound (reading input), shuffle-bound (moving data), skew-bound (one reducer doing all the work), or output-bound (writing results)? The answer picks the fix.

3 · Joins in batch: never do a per-record lookup

A join combines two datasets on a shared key: attach each event to the user who generated it, attach each click to the ad that was shown, attach a training label to the feature row it belongs to. The naive way — the way that feels obvious if you come from request-serving code — is: scan the big dataset, and for each record, look up the matching row in a database. This is a disaster at batch scale, and seeing exactly why is the heart of this lesson.

The batch answer is to bring both datasets into the job and join them with sequential access instead of random lookups. The strategies fall into two families, split by where the join happens. A reduce-side join does the matching after the shuffle, in the reducer, and works for any input sizes. A map-side join does the matching inside the mapper and skips the shuffle of the big side entirely — but only when the inputs have a shape that allows it. Which one wins depends on the relative sizes.

Reduce-side: the sort-merge join (the general case, both sides shuffled)

The workhorse reduce-side strategy is the sort-merge join, and it uses the shuffle itself. Map over both inputs — the billion events and the user profiles — and emit each record keyed by the join key (the user id). The shuffle partitions and sorts both sides by that key, so when a reducer receives a partition, it sees all the events for a given user and that user's profile record gathered and sorted right next to each other. The reducer walks the sorted run once, merging matching keys. No random lookups, no live database — just two big sequential scans, a sort, and a linear merge. The cost is one full shuffle of both datasets, but it scales: a billion events is a sequential sort over data laid out on disk, not a billion seeks. Where the naive version took ~11.6 days, a sort-merge over the same billion rows is a handful of sequential passes — hours, not days, and infinitely retryable because it touches only immutable input files.

One subtlety makes the merge clean: secondary sort. Within a single key the reducer wants the records to arrive in a controlled order — typically the one profile record first, then the user's many events — so it can read the profile into a variable and then attach it to each event in a single forward pass, never buffering the whole group. The framework arranges this by sorting not just on the join key but on a composite key (join key, then a small table tag that marks “profile” as sorting before “event”), while still partitioning on the join key alone so the group stays together. That extra ordering within a key is the secondary sort.

Map-side: broadcast and partitioned hash joins (skip the shuffle)

Map-side joins avoid the shuffle when the input shape allows it. The common case is the broadcast hash join: if one side is small enough to fit in memory — say the billion events join against a 2-million-row user table that is a few hundred MB — you do not shuffle the big side at all. Load the small side into an in-memory hash table and ship a copy of that hash table to every mapper; each mapper then scans its slice of the big input and probes the local hash table for each record — an in-memory hash lookup, nanoseconds, no network. The big side is never shuffled. This is dramatically faster than a sort-merge join when it applies; the constraint is purely that the broadcast side must fit in memory on every worker.

The other map-side case is the partitioned hash join (a map-side merge join): if both inputs are already partitioned the same way — same number of partitions, partitioned on the join key, e.g. because an earlier job laid them out that way — then partition i of one side can only match partition i of the other. A mapper handling partition i reads both sides' partition i locally and joins them with no shuffle and no broadcast, because the matching keys are already co-located. The win is real but it depends on the inputs being co-partitioned in advance; otherwise you are back to paying for the shuffle with a sort-merge join.

4 · Skew: one hot key wrecks one reducer

Lesson 11 warned about the hot key — a single partition key with vastly more data than the others, turning one node into a bottleneck. The shuffle inherits this problem exactly, because the shuffle is a partitioning. If you join events to users by user_id and one "user" is actually a shared service account that generated 30% of all events, then the reducer responsible for that key's partition has to process 30% of the data alone. Every other reducer finishes in minutes; that one runs for hours. The whole job's wall-clock time is the slowest reducer — a straggler — so a single hot key can make a 1,000-machine cluster as slow as one machine.

The fixes mirror lesson 11's hot-key remedies, and for joins they have specific names. The general technique is the skewed join (also called a sharded or replicated join): you spread the hot key's records across many reducers and replicate the other side of the join so each shard can still find its matches. Concretely — salt the hot key on the big side: append a random suffix (user_id#0 … user_id#9) so its records split across, say, 10 reducers instead of crashing into one. But now those 10 reducers each hold only a slice of the events for that user, and the single matching profile record must reach all of them — so you replicate the small/other side of the hot key to every shard (emit the profile as user_id#0 through user_id#9). Each shard then joins its slice of events against the replicated profile; a second pass combines the partial results if the output must be regrouped. You trade an extra stage and some duplication of the small side for real parallelism on the hot key. Map-side pre-aggregation (a combiner) is the cheaper fix when the join is followed by an aggregation: it reduces the volume before the shuffle so the hot reducer receives summaries, not raw rows. And if the hot side is small, a broadcast join sidesteps skew entirely, since there is no shuffle of the big side to concentrate. Modern engines also detect skew at runtime and split the offending partition automatically.

5 · Materialization, and why deterministic recompute is the whole point

A batch job writes its result down — to files, a table, an index. Materialization means storing a computed result so future reads are cheap, rather than recomputing it on every read. The output of one MapReduce stage is materialized to durable storage before the next stage reads it. This is the conservative choice MapReduce makes, and it has a famous downside (next), but it buys something precious: because every intermediate result sits on disk, a failed task can simply re-run.

Here is the crown jewel of batch processing. The input is an immutable snapshot, and the computation is a pure function of that input. Therefore the output is a deterministic recompute: re-run the job on the same input and you get bit-for-bit the same output. Three enormous consequences follow:

• Failures just retry. If a machine running a map or reduce task dies, the framework re-runs that one task on another machine from the same immutable input. The retried task produces the same output as the dead one would have — no coordination, no rollback, no "what state were we in?" The whole fault-tolerance story is "the task is a pure function; lose it, recompute it." Contrast this with the live database of Part II, where a failed write mid-transaction is a genuine coordination problem.
• You can fix a bug and reprocess history. Found a defect in last quarter's revenue calculation? Fix the code, re-run the job against the same archived input, and the corrected output replaces the wrong one. The source of truth was never mutated; the derived output is disposable and regenerable. This is impossible if your "computation" was a stream of irreversible in-place updates.
• Derived data is honest about being derived. A materialized index, training table, or feature snapshot is explicitly a function of an input plus a code version. Record both and the output is reproducible and auditable — exactly the property lesson 20 generalizes into the whole derived-data discipline.

The bill for all this is latency. You wait for all the input to exist (you cannot start until the snapshot is complete) and you wait for the whole job to finish before any output is usable. A nightly job's output is up to a day stale. Batch trades freshness for the simplicity and correctness of recompute — and lesson 19 is, in large part, the story of clawing back that latency.

Dataflow engines: keep MapReduce's recompute, cut its disk writes

Classic MapReduce materializes every stage's output to the distributed filesystem (often replicated 3×) before the next stage starts. For a multi-stage job that is a lot of slow, redundant disk I/O — and the next stage cannot begin until the previous one fully lands. Modern dataflow engines (Spark, Flink in batch mode, Tez) keep the same model — a directed graph of map/shuffle/reduce-like operators over immutable input — but optimize the execution: hold intermediate results in memory and pipeline them between stages instead of round-tripping every stage through disk, and plan the whole operator graph at once so they can fuse narrow operations and pick join strategies globally. They keep deterministic recompute for fault tolerance — Spark, for instance, tracks each partition's lineage (the operations that produced it) so a lost partition is rebuilt by replaying just those operations from the immutable input — while avoiding MapReduce's habit of writing every stage to disk.

Iterative graph processing: where plain MapReduce is the wrong tool

Some batch computations are not one pass but many rounds over the same data: PageRank, shortest paths, connected components, label propagation — algorithms that walk a graph, updating each vertex from its neighbors, and repeat until the values stop changing (converge). Plain MapReduce is poor at these. Each round is a separate MapReduce job, and a MapReduce job's only way to pass state to the next round is to write the entire dataset back to the distributed filesystem and read all of it again. So a 30-iteration PageRank re-reads and re-writes the whole graph 30 times, even though most vertices changed little between rounds — the I/O dwarfs the actual computation.

The purpose-built model is Pregel, an instance of the bulk synchronous parallel (BSP) model (Google's Pregel; open-source Apache Giraph; also the GraphX and GraphFrames APIs). The graph's vertices are partitioned across machines and kept in memory between rounds. Computation proceeds in supersteps: in each superstep every vertex runs the same small function — it reads the messages its neighbors sent it last round, updates its own value, and sends messages to its neighbors for the next round. A synchronization barrier separates supersteps (every message from round n is delivered before round n+1 begins — that is the “bulk synchronous” part), and a vertex that has nothing left to do votes to halt; when all vertices have halted, the job is done. Because the graph stays resident and only the (small) messages move, BSP avoids MapReduce's per-round full re-read — the difference between shipping a few updates and rewriting the whole graph every iteration. The same fault-tolerance idea still applies: periodic checkpoints of vertex state let a failed superstep replay from the last checkpoint rather than from scratch.

ML-infra tie: this is how you build a training set

Batch is the engine room of offline ML. Building a training set is a batch join: scan the event log (the labels — clicks, conversions, ratings), join each label to the feature values as of the time the event happened (a join by entity id and time), and materialize the result as a flat training table. Recompute it when the feature definition changes — fix how "7-day click rate" is computed, re-run over the archived event log, get a corrected training set, no production data touched. Precomputing features for a feature store, rebuilding an embedding index, generating offline evaluation metrics, backfilling a new feature across all history — every one of these is a bounded-input batch job whose superpower is that you can re-derive it. The "derived data" idea sketched here is exactly what lesson 20 turns into a discipline.

Checkpoint exercise

Where this points next

Batch's great strength — bounded, immutable input you can scan whole and recompute — is also its hard limit. Real systems produce data continuously: clicks, sensor readings, log lines, profile edits stream in forever, and waiting for "all the input" before you can produce any output means your derived data is always hours stale. The next lesson lifts the boundedness assumption. Stream processing (lesson 19) treats the input as an unbounded, ordered log of events, processes each as it arrives, and reframes batch's concepts for an endless world: the immutable input log becomes a change-data-capture (CDC) feed, the materialized output becomes a continuously-updated view, the shuffle becomes a stateful partitioned operator, and "wait for all the input" becomes the much subtler question of when is a window complete when events can arrive late. Many of the joins and aggregations you just learned reappear — now they must run forever.

Interview prompts

• What are the three phases of MapReduce and which one is expensive, and why? (§1, §2 — map (local, parallel, emits key-value), shuffle (partition by key + sort — the all-to-all that moves data across the network, so it dominates cost), reduce (aggregate all values per key). The shuffle is a repartition, lesson 11.)
• Why can't you join a billion-row log to a database with a per-record lookup? (§3 — ~1e9 × ~1 ms random I/O ≈ 11.6 days; it is random access not sequential, has no fault tolerance / clean resume, and overloads the live production store. Bring both sides into the job instead.)
• Contrast sort-merge and broadcast hash joins and say when each wins. (§3 — sort-merge shuffles and sorts both sides by the join key and merges per key, general at any size but pays a full shuffle; broadcast loads the small side into an in-memory hash table on every mapper and probes it, skipping the big-side shuffle, but only when the small side fits in memory.)
• A join job has 99 reducers done and 1 running for hours — what happened and how do you fix it? (§4 — a hot/skewed join key (lesson 11) concentrated most rows on one reducer (a straggler); fix by salting / splitting the hot key across reducers, map-side pre-aggregation, or broadcasting if that side is small.)
• Why does a batch task tolerate machine failure so simply? (§5 — the task is a pure function of an immutable input snapshot, so a lost task is just re-run on another machine and reproduces the same output — no rollback or coordination, unlike a live transactional write.)
• What is deterministic recompute and what two superpowers does it give you? (§5 — same immutable input + same code → same output; so failed tasks retry freely, and you can fix a bug and reprocess all history without ever mutating the source of truth. The cost is latency: you wait for all input and the whole job.)
• How do Spark/Flink-batch improve on classic MapReduce while keeping fault tolerance? (§5 — they keep the deterministic recompute model (Spark tracks partition lineage to rebuild lost data by replay) but hold intermediate results in memory and pipeline stages instead of writing every stage to the distributed filesystem.)
• How is building an ML training set a batch job? (§5 — scan the event log for labels, join each label to the feature values as of the event time, materialize a flat training table; recompute over the archived log when the feature definition changes — derived, reproducible data, lesson 20.)