Chapters 5 to 9 treated systems that answer requests online. Part III shifts to derived data: datasets computed from other datasets, where correctness is measured by throughput over a bounded input rather than by latency per request. A batch job reads a fixed input, runs for minutes or hours, and writes an output; nobody is waiting on the other end of a socket. That changes which trade-offs matter, and it makes fault tolerance much easier, because any failed piece can simply be redone.

The chapter starts with Unix pipelines, which embody the same ideas at single-machine scale, then builds MapReduce and its joins, then shows why the industry moved past MapReduce toward dataflow engines such as Spark and Flink. For interview purposes, the valuable parts are the join algorithms, the "write output atomically and keep inputs immutable" philosophy, and the reasons intermediate state on disk was replaced by in-memory pipelining.

In one minute

• Unix tools compose because they share a uniform interface (byte streams, line-separated records), separate logic from wiring (pipes), and treat inputs as immutable, so experiments are cheap and reversible.
• MapReduce is Unix pipelines scaled to thousands of machines: a mapper extracts key-value pairs, a shuffle sorts and partitions by key, a reducer processes all values for a key. Jobs are chained through files on a distributed filesystem (HDFS).
• HDFS uses shared-nothing commodity machines, replicating blocks for durability; MapReduce schedules mappers near their input blocks ("putting the computation near the data").
• Reduce-side joins (sort-merge) bring all records with a key to one reducer; they need no assumptions about input but pay the full shuffle. Hot keys need skew handling.
• Map-side joins avoid the shuffle when one input fits in memory (broadcast hash join) or both inputs are partitioned and sorted the same way (partitioned hash join, map-side merge join).
• Batch outputs are search indexes, key-value files for serving, or analytics tables; build them as immutable files and swap them in, never mutate a live database row by row from a job.
• Hadoop differs from MPP databases by storing raw data first (schema-on-read, the "data lake"), supporting arbitrary code, and tolerating individual task failures instead of restarting the query.
• MapReduce's weakness is materializing every intermediate result to disk. Dataflow engines (Spark, Tez, Flink) express a whole job as a graph of operators, pipeline data between them, and recompute lost partitions from lineage.
• Graph algorithms use Pregel/BSP: iterate until convergence, with vertices exchanging messages each superstep.
• High-level APIs (Spark SQL, Hive, DataFrames) let batch systems borrow query optimisation from databases, while databases borrow flexibility from batch; the two are converging.

Batch processing with Unix tools

A log analysis in one line

Suppose you want the five most requested URLs from a web server log. A chain of awk to extract the URL, sort to group identical lines, uniq -c to count, sort -r -n to rank and head -n 5 to cut gives the answer with a few lines of shell. Written in Ruby or Python, the same job is a hash map from URL to count followed by a sort of the hash entries.

The two approaches differ in where the state lives. The hash map must fit in memory, which is fine for a day of logs on one server. The sort approach does not need it to: GNU sort spills to disk and sorts chunks that are merged afterwards, and it uses multiple cores. The "sort then count adjacent runs" pattern scales to inputs far bigger than memory with no extra engineering. That same pattern is the heart of MapReduce.

The Unix philosophy

Doug McIlroy's idea of connecting programs like garden hose became pipes in 1973. The design rules matter because MapReduce copies them:

• Do one thing well. Small programs that are composed rather than one monolith.
• A uniform interface. Every program reads and writes a byte stream, conventionally lines of ASCII text with a separator character. Because the interface is identical, any output can become any input. HTTP URLs are the only comparably successful uniform interface since.
• Separation of logic and wiring. A program reads stdin and writes stdout and does not care whether they are files, pipes or terminals. Loose coupling or inversion of control: the shell decides the wiring.
• Transparency and experimentation. Inputs are immutable files, so a failed run does no damage; you can stop a pipeline at any point and inspect the intermediate output with less; you can write one stage's output to a file and feed it to the next stage later.

The limitation is that it runs on one machine. That is where Hadoop comes in.

MapReduce and distributed filesystems

Distributed filesystem

MapReduce jobs read and write files on a distributed filesystem. HDFS (Hadoop Distributed File System) is the open-source version of Google's GFS; others are GlusterFS, QFS and object stores like Amazon S3 used in the same role. HDFS follows the shared-nothing principle: ordinary machines each run a DataNode daemon exposing their local disks over the network, and a central NameNode tracks which blocks live on which machine. Files are split into blocks (typically 64 to 256 MB) that are replicated on several machines or protected by erasure coding (Reed-Solomon), so that losing a disk or a node loses no data. Compared with a shared SAN appliance, this is dramatically cheaper and scales to tens of thousands of machines and hundreds of petabytes.

The MapReduce job

A job has the shape of the Unix example:

1. Read the input files and break them into records (lines, in the log case).
2. Mapper: a function called once per record, emitting any number of key-value pairs. It keeps no state between records. In the example it extracts the URL as the key with an empty value.
3. Shuffle: sort all pairs by key, so equal keys are adjacent.
4. Reducer: a function called once per distinct key with an iterator over that key's values, emitting output records. In the example it counts the values.

The mapper does the equivalent of awk, the reducer of uniq -c, and the framework does the sort. A second MapReduce job sorts by count and takes the top five.

Distributed execution

Parallelism comes from partitioning. The scheduler assigns one map task per input block and tries to run it on a machine holding a replica of that block, so the input is read from local disk rather than shipped across the network. The mapper code (a JAR or a native binary) is copied to those machines; moving code is cheaper than moving data.

The number of reduce tasks is configured by the job author. Each mapper partitions its output by a hash of the key, sorts each partition and writes it to local disk. When a mapper finishes, the scheduler tells the reducers to fetch their partitions over the network, which is the shuffle. Each reducer merges its sorted inputs from all mappers (a merge sort keeps memory bounded), then calls the user's reduce function for each key with its values. Output goes to the distributed filesystem, usually one file per reducer, replicated as normal.

Workflows

Real analyses chain jobs: one job's output directory is the next job's input. MapReduce has no built-in support for that, so workflow schedulers (Oozie, Azkaban, Luigi, Airflow) manage dependencies, and higher-level tools (Pig, Hive, Cascading, Crunch) compile a program into a chain of jobs. Workflows of fifty to a hundred jobs are normal for a recommendation system.

Joins and grouping

Joins are where batch jobs do most of their work. A database would use an index to find the matching rows for one record; MapReduce has no indexes and scans the full input every time. That is fine when the query touches a large fraction of the data, which is what batch jobs do. Because requests to a remote database inside a mapper would be slow, nondeterministic and a load on the database, the convention is to take a copy of the database (via an ETL dump) into HDFS and join locally.

Sort-merge join

Take a user activity log (events keyed by user ID) and a user database (user ID to date of birth), and compute activity by age bracket. Two sets of mappers run: one emits (user ID, event) from the log, the other emits (user ID, date of birth) from the user table. The framework sorts both by user ID, so a reducer sees all records for one user together. A secondary sort places the user record first among its key's values, so the reducer reads the birth date once and then streams the events, emitting (age bracket, event) pairs without buffering. This is a sort-merge join: mappers emit, the shuffle merges by key, the reducer joins.

Because the shuffle brings everything for a key to one place, the reducer's logic never touches the network. The job author does not reason about where data lives; MapReduce separates network communication (the shuffle) from application logic (map and reduce), which also simplifies failure handling.

Group-by

The same mechanism does GROUP BY: set the key to the grouping attribute and the reducer sees each group. Counting records, summing fields, picking the top k, or sessionization (collecting all events of one user session to see whether a UI change led to more purchases) all fit this shape. Grouping and joining are implemented identically.

Handling skew: hot keys

If one key has far more records than the others (a celebrity's follower list in a social graph, a wildly popular product), the reducer for that key becomes a straggler: a MapReduce job finishes only when its slowest reducer does. Fixes:

• Pig's skewed join samples the input to find hot keys, then spreads a hot key's records across several reducers at random while replicating the other side of the join to all of them.
• Hive's skewed join needs hot keys declared in the table metadata and uses a map-side join (below) for those keys.
• A two-stage aggregation: group the hot key's records in a first stage spread across reducers with a random suffix, then combine the partial results in a second job.

Map-side joins

A reduce-side join needs no assumptions about the inputs but pays for sorting, copying to reducers and merging. If you know something about the data, a mapper can do the join alone with no reducers and no shuffle.

Technique	Requirement	How it works
Broadcast hash join	One input small enough to fit in each mapper's memory	Each mapper loads the small input into a hash table, then streams the large input and probes for matches. Used by Pig (replicated join), Hive (MapJoin), Cascading, Crunch; Impala does the same at query time
Partitioned hash join	Both inputs partitioned the same way (same number of partitions, same key, same hash)	Mapper for partition 3 of the log loads only partition 3 of the users into a hash table, so each hash table is small. Hive calls it a bucketed map join
Map-side merge join	Both inputs partitioned and sorted by the join key	The mapper merges two sorted streams the way a reducer would; no hash table needed. Natural when the inputs are outputs of earlier jobs with the same grouping

The requirement in each row is why these are not free: a map-side join depends on the physical layout (partitioning, sort order, size) of its inputs, so that layout becomes part of the contract between jobs. Workflow metadata (HCatalog, Hive metastore) exists partly to record it. Output also differs: a reduce-side join produces output partitioned and sorted by the join key; a map-side join's output follows the large input's partitioning.

The output of batch workflows

Batch jobs are neither OLTP nor analytics in the usual sense; they are closer to analytics in scanning big inputs, but their output is often not a report but another dataset.

Search indexes

Google originally used MapReduce to build its web index, and Lucene/Solr indexes are still built this way. The index is a dictionary from term to document IDs; building it is a parallelisable, partitioned batch job over the document set. Because index files are immutable once written, the job writes new index segments to the filesystem and the search servers pick them up; to update the index, rebuild it periodically or add incremental segments and merge in the background.

Key-value stores as output

Machine learning pipelines (classifiers, recommendation systems) emit a database of results: for each user ID, a list of suggested friends or products, served by an online system. The naive approach, having reducers issue writes to a database over the network, is wrong for several reasons. The per-record round trip is slow compared with batch throughput; a parallel job hammers the database and degrades the production service; and if a job fails halfway, the database holds half-new, half-old data, destroying the clean "retry the whole thing" property of batch jobs.

The right pattern is to build a complete immutable database file inside the batch job and then bulk-load it into read-only serving stores. Voldemort, Terrapin, ElephantDB and HBase bulk loading all do this: the job writes the files to HDFS, the serving nodes copy them, switch over atomically, and keep the old files to roll back if anything is wrong.

The philosophy of batch outputs

Inputs are immutable, outputs are written to a new location, and nothing is modified in place, so:

• Bugs are recoverable: fix the code, rerun, and the old output is gone. Databases with destructive updates cannot offer this, a property often called human fault tolerance.
• Retries are safe: a failed job can be rerun from scratch because its partial output is discarded, which is why MapReduce can treat task failure casually.
• One dataset can feed many jobs, including exploratory ones, with no risk.
• Logic and wiring separate as in Unix: a team writes the mapper, another decides where its output goes.

Unlike Unix, Hadoop uses structured formats (Avro, Parquet) instead of ad-hoc text parsing, which removes the most tedious part of shell pipelines.

Comparing Hadoop to distributed databases

Massively parallel processing (MPP) databases such as Teradata, Greenplum and Vertica also ran parallel joins and aggregations across a cluster of shared-nothing machines long before Hadoop. The differences are what made Hadoop interesting:

Aspect	MPP database	Hadoop (HDFS + MapReduce)
Data model	Schema designed up front; data imported and indexed	Any bytes; raw dumps, text, images; schema imposed on read
Processing	SQL only	Arbitrary code (any language) alongside SQL via Hive
Workload	Analytic queries returning fast	Long-running ETL, ML training, index builds
Fault handling	A failed node aborts the whole query; retry everything	Individual task retried; job continues
Resource use	Dedicated cluster, memory-resident	Shares machines with other jobs; favours disk, tolerates preemption

Diversity of storage. Hadoop encourages dumping raw data first and deciding how to process it later, the sushi principle ("raw data is better"), which the industry named the data lake. Different teams can derive their own views. This shifts the burden of interpreting data to the consumer (schema-on-read), which is the right default when the producers and consumers are different teams.

Diversity of processing. SQL is not the only thing people want to do with data. Building indexes, training models, running image processing or graph algorithms do not fit a query language, and HDFS lets all of them share one storage system.

Designing for frequent faults. MapReduce reruns individual map or reduce tasks on failure, and writes intermediate results to disk so a partially complete job survives. That looks like over-engineering until you know the context: at Google, batch jobs ran at low priority on shared clusters and were routinely preempted when a higher-priority online service needed the machines, so a map task had a meaningful chance of being killed mid-run. An MPP database restarting the whole query on every preemption would never finish. In clusters where preemption is rare, the design is less justified, which is part of the story of what came next.

Beyond MapReduce

MapReduce is simple to understand but hard to use directly; higher-level tools (Pig, Hive, Cascading, Crunch) made it tolerable. Its deeper issue is performance of the execution model itself.

The cost of materialization

Every MapReduce job writes its full output to HDFS before the next job can read it. This materialization of intermediate state has costs:

• A job can only start when all tasks of the preceding job finish, so a single straggler delays the whole workflow. Unix pipes, by contrast, stream output to the next process as it is produced.
• Mappers often just reread the file the previous reducer wrote and re-partition it; the mapper stage is redundant when the previous reducer could have partitioned its output directly.
• Replicating intermediate data across HDFS nodes is wasteful for data that only exists for a few minutes.

Dataflow engines

Spark, Tez and Flink (and Google's Dryad and Nephele earlier) fix this by handling an entire workflow as one job. Instead of alternating map and reduce stages, the job is a directed acyclic graph of operators; data flows along edges between them. The engine can:

• Sort only where sorting is needed, instead of between every stage.
• Pipeline records from one operator straight into the next without writing to disk, like a Unix pipe, and avoid redundant map stages.
• Use the knowledge of the whole graph to place operators that join on the same key on the same machine, and to choose join algorithms (hash, broadcast, merge) per operator.
• Reuse JVM processes across operators instead of launching a new process per task.

This is why Spark reported order-of-magnitude speedups over Hadoop MapReduce on iterative workloads. Pig, Hive and Cascading can all compile to Tez or Spark instead of MapReduce without changing user code.

Fault tolerance without materialization

If intermediate state is not on HDFS, a failed node loses it. Dataflow engines recompute it instead. Spark tracks the lineage of each RDD (resilient distributed dataset): which input partitions and which deterministic operators produced it. Flink checkpoints operator state periodically. On failure the engine recomputes only the lost partitions from the nearest surviving data.

Recomputation is only correct if operators are deterministic: the same input must give the same output, otherwise a recomputed partition disagrees with downstream consumers that already saw the first version. Random numbers, hash codes depending on object identity, wall-clock time and iteration order of hash tables all break this, and the engine's answer is to kill downstream operators too. When intermediate data is much smaller than its inputs (a filter) or expensive to recompute (a long CPU-bound stage), materializing it to disk is still the right call, so engines leave the choice to the author.

Graphs and iterative processing

Graph algorithms such as PageRank, shortest paths or transitive closure iterate until convergence, which a DAG of operators cannot express directly. Running one MapReduce job per iteration works but is slow, since each iteration rereads the whole graph. The Pregel model (Google, implemented by Apache Giraph, Spark GraphX, Flink Gelly) is the bulk synchronous parallel (BSP) approach:

1. Each vertex has state and a user-defined function.
2. In each superstep, the function runs for every vertex that received messages in the previous superstep, updates the vertex's state and sends messages to other vertices along edges.
3. The framework delivers all messages before the next superstep starts; messages are batched, so it is efficient.
4. The algorithm ends when no vertex has messages to send.

Vertex state persists across supersteps, so only changed vertices do work. Fault tolerance is by periodic checkpoints of all vertex state at superstep boundaries, with rollback and replay on failure. The weakness is communication: unless the graph partitions cleanly (it rarely does), most messages cross machine boundaries. For a graph that fits on a single large machine, a single-node algorithm (GraphChi, or simply loading it into memory) is often faster than a cluster.

High-level APIs and the convergence with databases

Hive, Pig, Cascading and Crunch grew on MapReduce; Spark's DataFrames and SQL, Flink's Table API and Tez have made declarative, relational-style programming the norm on dataflow engines. Writing joins declaratively lets the engine choose the algorithm (cost-based optimisation), reorder filters, and, with columnar formats such as Parquet, read only needed columns and vectorise the work. These are the techniques MPP databases had for decades.

In the other direction, databases gained user-defined functions, JSON and columnar storage, and Hadoop gained SQL. The convergence means a modern batch system combines database-style query optimisation and storage formats with Hadoop's flexibility: arbitrary code when you need it (model training, custom parsers) and declarative queries when you do not. Specialised libraries sit on top for ML (MLlib), graphs (GraphX) and statistics, mirroring the way stored procedures never replaced application code.

Design trade-offs

Decision	Choose this	When
Join strategy	Reduce-side sort-merge	No layout assumptions; inputs large and unsorted; output must be sorted by key
	Broadcast hash	One side fits in memory (dimension tables, lookup data)
	Partitioned hash or merge	Inputs already co-partitioned, typically as outputs of earlier jobs
Intermediate state	Materialize to HDFS	Expensive to recompute, much smaller than input, or consumed by many downstream jobs
	Pipeline in memory	Short-lived, cheap to recompute, deterministic operators
Output	Immutable files, bulk loaded	Serving data (recommendations, indexes); you want rollback and safe retries
	Direct writes to a DB	Almost never from a batch job; the load and partial failure cost are too high
Engine	MapReduce	Legacy; preemption-heavy clusters; teaching
	Spark / Flink / Tez	Nearly all new batch work; iterative and multi-stage workflows
	Pregel-style BSP	Graph algorithms over graphs too large for one machine
Storage	Data lake (raw dumps, schema-on-read)	Many consumers, uncertain future uses, non-relational data
	MPP warehouse	Known schema, interactive SQL, strict SLAs on query latency

Interview angle

• Explain MapReduce to someone who knows Unix pipes. Mapper is awk (extract a key per record), the shuffle is sort, the reducer is uniq -c or any per-group function. The framework parallelises each stage across machines by partitioning the input into blocks and the keys by hash, and writes results to a distributed filesystem the way a shell writes to a file. Jobs chain like pipeline stages, but via files, not streaming pipes.

• How would you join a 10 TB clickstream with a 2 GB user table? Broadcast hash join: ship the user table to every mapper, load it in a hash table, stream the clickstream and probe. No shuffle of the 10 TB. If the user table were 500 GB, use a reduce-side sort-merge join, or if both are already partitioned by user ID the same way, a partitioned hash join per partition.

• What is a hot key problem and how do you fix it? One key with disproportionate volume makes one reducer a straggler. Sample to detect hot keys, then spread each hot key's records across several reducers (random suffix) while replicating the other side of the join, and combine partial results in a second stage. Pig and Hive have built-in skewed joins.

• Why not have reducers write recommendations straight into the serving database? Row-at-a-time writes are slow relative to batch throughput, a parallel job overloads the production database, and a mid-job failure leaves a mix of old and new rows, destroying safe retry. Write complete immutable files and bulk-load them with an atomic switch and the previous version kept for rollback.

• Why was MapReduce's write-to-disk design reasonable at Google but a liability elsewhere? Google's batch jobs ran preemptible on shared clusters, so tasks were frequently killed; materializing intermediate output let jobs survive. In dedicated clusters preemption is rare, so the disk round trips and the wait for whole stages to finish are pure overhead, which dataflow engines removed by pipelining in memory and recomputing lost partitions from lineage.

• How does Spark recover from a lost executor without writing intermediate data to HDFS? Each RDD partition records its lineage (parent partitions and the deterministic operator applied). The lost partition is recomputed from surviving ancestors, or from the input on HDFS. Operators must be deterministic; nondeterministic ones force downstream recomputation too. Flink reaches the same goal with periodic checkpoints of operator state.

• Design a daily pipeline that builds a search index for 1 billion documents. Store documents in object storage or HDFS; a map stage parses and tokenises, a reduce stage groups postings by term and writes immutable index segments partitioned by document range. Build into a new directory, validate, then atomically point serving nodes at it, keeping the previous build for rollback. For fresher updates, add incremental segments between daily rebuilds and merge in the background.

Pitfalls and misconceptions

• "MapReduce is Hadoop." MapReduce is one programming model; Hadoop is the ecosystem (HDFS, YARN, Hive, and now mostly Spark or Tez as engines). New code rarely targets raw MapReduce.
• Thinking batch systems have indexes. They scan everything. That is efficient for queries touching most of the data and terrible for point lookups, which belong in a serving store built by the batch job.
• Querying a live database from inside mappers. It is slow, nondeterministic (the database changes under you), and it risks taking down production. Copy the data into the batch system first.
• Ignoring determinism when relying on lineage recomputation. Random seeds, time, and hash-order iteration make recomputed partitions diverge from the originals.
• Assuming reduce-side joins are always the fallback. They are the general case, but on co-partitioned inputs they waste a full shuffle; record and exploit partitioning in workflow metadata.
• Mutating outputs in place. You lose rollback and safe retry, the two properties that make batch processing forgiving.

Key terms

Term	Meaning
Derived data	A dataset computed from another dataset, which can be rebuilt from its source
Uniform interface	Unix convention that every program consumes and produces byte streams, making arbitrary composition possible
HDFS	Hadoop's distributed filesystem: shared-nothing nodes, replicated or erasure-coded blocks, a NameNode for metadata
Mapper	Stateless function emitting key-value pairs per input record
Reducer	Function called once per key with all of that key's values, after the shuffle
Shuffle	Partitioning mapper output by key hash, sorting, and copying to reducers
Sort-merge join	Reduce-side join in which both inputs are emitted with the join key and the reducer joins adjacent records
Secondary sort	Ordering values within a key so a reducer sees, for example, the user record before the user's events
Hot key / skew	A key with far more records than others, making one reducer a straggler
Broadcast hash join	Map-side join that loads the small input into memory on every mapper
Partitioned hash join	Map-side join on inputs partitioned identically, loading only the matching partition
Map-side merge join	Map-side join on inputs partitioned and sorted identically, merging two sorted streams
Bulk loading	Building immutable database files in the job and switching serving nodes to them atomically
Data lake / sushi principle	Store raw data first and impose schema on read
Materialization	Writing an intermediate result fully to storage before the next stage reads it
Dataflow engine	Executes a whole job as a DAG of operators with in-memory pipelining (Spark, Tez, Flink)
Lineage	The record of inputs and operators that produced a partition, used to recompute it after failure
Pregel / BSP	Vertex-centric iterative graph processing in synchronous supersteps exchanging messages
Cost-based optimisation	Choosing join algorithms and operator order using statistics about the data

Further reading

• DDIA, Chapter 10: "Batch Processing"; Kleppmann's references at https://github.com/ept/ddia-references
• Jeffrey Dean and Sanjay Ghemawat, "MapReduce: Simplified Data Processing on Large Clusters," OSDI 2004: https://research.google/pubs/pub62/
• Sanjay Ghemawat, Howard Gobioff and Shun-Tak Leung, "The Google File System," SOSP 2003
• Matei Zaharia et al., "Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing," NSDI 2012: https://www.usenix.org/conference/nsdi12/technical-sessions/presentation/zaharia
• Grzegorz Malewicz et al., "Pregel: A System for Large-Scale Graph Processing," SIGMOD 2010
• Michael Stonebraker et al., "MapReduce and Parallel DBMSs: Friends or Foes?," CACM, 2010
• Adam Drake, "Command-line Tools can be 235x Faster than your Hadoop Cluster," 2014
• Nathan Marz, "How to beat the CAP theorem" (the human-fault-tolerance argument for immutable batch outputs), 2011