MapReduce: Simplified Data Processing on Large Clusters

In one breath — the whole paper, compressed

Google engineers had written hundreds of one-off distributed programs over crawled documents and request logs, and every one of them re-implemented partitioning, scheduling and failure handling. MapReduce reduces that to two user functions: a map that turns each input record into intermediate key/value pairs, and a reduce that merges all values sharing a key. A master process splits the input into M pieces, assigns map and reduce tasks to idle workers, tracks where each map task left its intermediate files on local disk, and re-executes any task whose worker stops answering pings. Because the model is restricted and the operators are usually deterministic, re-execution alone is a sufficient recovery mechanism, and the same restriction lets the master place tasks next to their data and launch redundant backup copies of stragglers. On roughly 1800 machines the system grepped about a terabyte in some 150 seconds and sorted a terabyte in 891 seconds, beating the best then-reported TeraSort time of 1057 seconds.

Before this paper — the world it landed in

By 2003 Google was running hundreds of special-purpose computations over crawled documents, web request logs and web-graph data to produce inverted indices, per-host page counts and summaries of the day's most frequent queries. Each computation was conceptually straightforward, but the input was large enough that it had to be spread over hundreds or thousands of machines, so each program was hand-written with its own partitioning, work distribution and failure-recovery code that buried the simple computation underneath. The hardware made this harder rather than easier: dual-processor commodity x86 Linux machines with 2-4 GB of memory and cheap IDE disks, wired at 100 Mb/s to 1 Gb/s per machine but with far lower aggregate bisection bandwidth, in clusters big enough that machine failure was a routine event rather than an exception. The parallel abstractions available at the time, such as MPI, Bulk Synchronous Programming and parallel prefix models, raised the level of expression but had mostly been implemented at much smaller scale and left machine failures to the programmer. GFS, published one year earlier, had just made it reasonable to assume a replicated file system spread across those same local disks, and that is the substrate MapReduce is built on.

The problem — what was actually breaking

The raw input for these computations is large enough that the work has to be distributed across hundreds or thousands of machines simply to finish in a reasonable amount of time.
The code needed to parallelize the computation, distribute the data and handle failures conspires to obscure the original simple computation with large amounts of complex code.
A cluster of hundreds or thousands of commodity machines makes machine failure common, so a library that runs for hours must tolerate losing workers gracefully rather than aborting the job.
Network bandwidth is a relatively scarce resource in this environment: per-machine links are 100 Mb/s to 1 Gb/s, but overall bisection bandwidth averages considerably less, so naively shipping intermediate data around does not scale.
A single straggler machine can dominate total job time, whether from a bad disk that drops read performance from 30 MB/s to 1 MB/s, competition from other tasks on the machine, or a machine-initialization bug that disabled processor caches and slowed affected machines by over a factor of one hundred.
Programmers with no parallel or distributed systems experience could not exploit the cluster at all, and deterministic crashes in user code on a few bad records could prevent an entire operation from ever completing.

Core ideas — the contributions, and why they work

A deliberately restricted model

The user supplies only two functions, typed map (k1,v1) to list(k2,v2) and reduce (k2,list(v2)) to list(v2), and the library owns everything else. The restriction is the whole point: because every record is processed independently and every key's values are combined independently, the runtime is free to choose how many machines to use, where to run each piece, and when to run a piece twice. The values for a key reach the reduce function through an iterator rather than a list, so a single key may have more values than fit in memory. This is a simplification and distillation of earlier restricted parallel models, but with a fault-tolerant implementation that scales to thousands of processors.

Re-execution as the primary fault tolerance

There is no checkpointing of user state and no message logging; recovery is simply running a task again. The master pings every worker, marks a silent worker failed, and returns its tasks to the idle state for rescheduling. Completed map tasks must be re-run because their output sits on the local disk of the dead machine, while completed reduce tasks need no re-run because their output already lives in the global file system. This works only because the model makes tasks side-effect-free and re-runnable, which is precisely what the restricted interface guarantees.

Move the computation to the data

Input files live in GFS as 64 MB blocks with typically three replicas on different machines, and the master reads that location information when assigning map tasks. It tries to run a map task on a machine that already holds a replica of its input split, and failing that on a machine on the same network switch as a replica. When a large MapReduce runs on a significant fraction of a cluster, most input is therefore read from local disk and consumes no network bandwidth at all. The idea is borrowed from active disks work, but applied to commodity machines with a few directly attached disks rather than to disk controller processors.

Backup tasks against stragglers

When an operation is close to completion, the master schedules backup executions of every task still in progress, and the task counts as done as soon as either copy finishes. This converts the tail of the job from a wait on the slowest machine into a race, without needing to diagnose why a machine is slow. The mechanism is tuned so it typically costs no more than a few percent extra computational resources. The measured payoff is large: the terabyte sort takes 44 percent longer when backup tasks are disabled.

Fine-grained tasks for dynamic balancing

M and R are chosen to be much larger than the number of worker machines, so each worker runs many tasks over the life of a job. Faster machines simply pull more tasks, which gives dynamic load balancing for free, and when a machine dies the many map tasks it had completed can be spread across all the remaining workers instead of being redone in one place. A typical production configuration is M = 200,000 and R = 5,000 on 2,000 worker machines. M is usually picked so each map task covers 16 MB to 64 MB, which is also what makes the locality optimization effective.

Combiners to shrink the shuffle

When the reduce function is commutative and associative, the user can register a combiner that runs on the map machine and partially merges intermediate records before they cross the network. Word counting is the canonical case: because word frequencies follow a Zipf distribution, a single map task emits thousands of identical records of the form the,1 that would otherwise all be shipped to one reduce task. Typically the same code implements both the combiner and the reducer, the only difference being that combiner output goes to an intermediate file destined for a reduce task rather than to the final output file. Partial combining significantly speeds up certain classes of MapReduce operations.

How it works — the mechanism, concretely

Splitting and task assignment

The library in the user program first splits the input files into M pieces, typically 16 MB to 64 MB each and controllable by an optional parameter, then starts many copies of the same program across the cluster. One copy is special, the master; the rest are workers. There are M map tasks and R reduce tasks to hand out, and the master repeatedly picks an idle worker and assigns it one of them. R and the partitioning function over the intermediate key space, by default hash(key) mod R, are specified by the user.

Map phase and the local spill

A worker assigned a map task reads its input split, parses key/value pairs out of it and passes each pair to the user Map function, buffering the emitted intermediate pairs in memory. Periodically the buffer is written to local disk, partitioned into R regions by the partitioning function, so each map task leaves behind R file regions, one per reduce task. The worker reports the on-disk locations of those regions back to the master. Intermediate data never goes to the replicated file system, which is exactly what saves network bandwidth.

Shuffle, sort and the reduce phase

The master forwards intermediate file locations to reduce workers, which use remote procedure calls to pull the relevant regions from the local disks of the map workers. Once a reduce worker has read all of its intermediate data it sorts it by intermediate key so that identical keys are adjacent, falling back to an external sort when the data does not fit in memory; the sort is needed because many distinct keys land in the same reduce task. The worker then walks the sorted run and, for each unique key, calls the user Reduce function with that key and an iterator over its values, appending the result to the final output file for that partition. Because a partition is processed in increasing key order, each output file is sorted, which is what makes distributed sort expressible and makes output usable for random-access lookup by key.

Master data structures and bookkeeping

For every map and reduce task the master stores its state (idle, in-progress or completed) and, for non-idle tasks, the identity of the worker executing it. For every completed map task it stores the locations and sizes of that task's R intermediate file regions, and pushes updates incrementally to workers with in-progress reduce tasks; the master is the conduit through which locations flow from map side to reduce side. This costs O(M + R) scheduling decisions and O(M * R) memory, though the constant is small at roughly one byte per map-task/reduce-task pair. When all tasks complete the master wakes the user program and the MapReduce call returns.

Failure detection and recovery

The master pings every worker periodically and marks it failed if no response arrives within a certain time; both completed map tasks and any in-progress map or reduce task on that worker are reset to idle and become eligible for rescheduling. When a map task first run on worker A is later re-run on worker B, every reduce worker is notified, and any reduce task that has not yet read A's data reads it from B instead. The scheme survives correlated loss: during one production run, network maintenance made groups of 80 machines unreachable for several minutes at a time and the master just re-executed their work and kept making forward progress. Master failure is the exception: checkpointing the master state would be easy, but since there is only one master and its failure is unlikely, the implementation simply aborts the computation and lets the client retry.

Atomic commit and failure semantics

Each in-progress task writes to private temporary files: a map task produces R of them, a reduce task produces one. On completing a map task the worker sends the names of its R temporary files to the master, which ignores the message if that map task is already recorded complete, so duplicate executions cannot corrupt the location table. A completing reduce worker atomically renames its temporary file to the final output file, relying on the file system's atomic rename so that if the same reduce task ran on several machines the final state contains exactly one execution's data. With deterministic operators the distributed run produces the same output as a non-faulting sequential execution; with non-deterministic operators the guarantee weakens to per-reduce-task equivalence, because the committed execution of R1 and of R2 may have read output from different executions of the same map task M.

Operational refinements

An optional mode skips records that crash user code deterministically: each worker installs a handler for segmentation violations and bus errors, records the sequence number of the current argument in a global variable, and on a signal sends a last-gasp UDP packet with that sequence number to the master, which tells the next re-execution to skip any record that has already failed more than once. For debugging there is an alternative library implementation that runs the whole job sequentially on one machine so gdb and profilers can be used. The master runs an internal HTTP server exporting status pages with task progress, byte counts, processing rates, links to per-task stderr and stdout, and which workers failed on which tasks. A counter facility lets user code count arbitrary events; values ride back on ping responses and the master de-duplicates counts from backup and re-executed tasks so nothing is double-counted.

What the paper showed — measurements and proofs

The benchmarks ran on roughly 1800 machines, each with two 2 GHz Intel Xeons with HyperThreading enabled, 4 GB of memory of which 1-1.5 GB was reserved by other tasks, two 160 GB IDE disks and a gigabit Ethernet link, arranged in a two-level tree-shaped switched network with about 100-200 Gbps of aggregate bandwidth at the root and sub-millisecond round-trip times.
Distributed grep over 10^10 100-byte records, searching for a three-character pattern that occurs in 92,337 records, with M = 15000 and R = 1, peaked at over 30 GB/s of input scanned once 1764 workers were assigned and finished in about 150 seconds, of which roughly a minute was startup overhead from propagating the binary and opening the 1000 input files in GFS.
Sorting 10^10 100-byte records (about 1 TB) took 891 seconds including startup, against the best then-reported TeraSort result of 1057 seconds, using fewer than 50 lines of user code with M = 15000 and R = 4000; input peaked near 13 GB/s, shuffling completed around 600 seconds and writes around 850 seconds, with 2 TB written because the output is 2-way replicated in GFS.
Disabling backup tasks on that same sort left all but 5 reduce tasks done at 960 seconds while the remaining stragglers took another 300 seconds, stretching the job to 1283 seconds, a 44 percent increase in elapsed time.
Intentionally killing 200 of 1746 worker processes several minutes into the sort caused a visibly negative input rate as completed map work was redone, yet the job still finished in 933 seconds, only 5 percent above the normal run.
In August 2004 Google ran 29,423 MapReduce jobs with an average completion time of 634 seconds, consuming 79,186 machine-days, reading 3,288 TB of input, producing 758 TB of intermediate data and writing 193 TB of output, averaging 157 worker machines and 1.2 worker deaths per job; rewriting the production indexing system on MapReduce cut one phase from about 3800 lines of C++ to about 700.

Limits and trade-offs — conceded and discovered

Conceded by the paper: there is a single master and no failover. Checkpointing its data structures would be easy, but because failure of one machine is unlikely the implementation simply aborts the whole computation if the master dies and asks clients to retry.
Conceded by the paper: M and R cannot grow without bound, since the master makes O(M + R) scheduling decisions and holds O(M * R) state in memory, and R is further constrained by users because every reduce task produces its own output file.
Conceded by the paper: the strong semantics hold only for deterministic operators. Non-deterministic map or reduce functions get the weaker guarantee that each reduce task individually matches some sequential execution, and there is no atomic two-phase commit across the multiple output files of one task, so tasks with cross-file consistency requirements must be deterministic.
Conceded by the paper: materializing intermediate data costs real I/O and bandwidth. Sort map tasks spend roughly half their time and I/O bandwidth writing intermediate output to local disk, and the output phase writes two replicas, which the paper notes erasure coding could reduce.
Exposed by later work: the rigid map-then-shuffle-then-reduce shape with disk materialization at every boundary is slow for iterative and interactive workloads, and the two-function interface is too low-level for multi-stage pipelines. Dryad generalized the shape to arbitrary dataflow graphs, Spark replaced the disk chain with lineage-tracked in-memory RDDs, and Pig, Hive, FlumeJava and DryadLINQ layered higher-level languages on top.

What it became — the systems that inherited it

MapReduce became the template for a decade of batch processing. Hadoop MapReduce reimplemented it almost feature for feature on HDFS, a GFS clone, including speculative execution for stragglers and locality-aware task placement, and turned the model into the default open-source big-data platform. On top of it, Hive and Pig compiled SQL-like and dataflow languages down to chains of MapReduce jobs, while FlumeJava, Cascading and DryadLINQ hid the job chaining behind pipeline APIs, tacitly conceding that raw map and reduce are too low-level for multi-stage work. Dryad generalized the fixed two-stage shape into arbitrary dataflow DAGs, and Spark kept the deterministic re-execution recovery model but tracked lineage over in-memory RDDs instead of materializing every stage to disk, which is what finally made iterative machine learning and interactive queries practical on the same clusters. Google itself moved past MapReduce for its original use case, replacing batch index rebuilds with incremental Percolator updates and later folding batch and streaming together in FlumeJava, Dataflow and Apache Beam. The durable inheritance is not the two-function API but the operational doctrine: locality-aware scheduling, fine-grained tasks for dynamic load balancing, speculative backup execution for the tail, and deterministic re-execution as the universal recovery mechanism, all still visible in YARN, Tez, Spark and Flink.

In the paper’s words — verbatim

“Users specify a map function that processes a key/value pair to generate a set of intermediate key/value pairs, and a reduce function that merges all intermediate values associated with the same intermediate key.”

Abstract

“Our use of a functional model with user-specified map and reduce operations allows us to parallelize large computations easily and to use re-execution as the primary mechanism for fault tolerance.”

§1

“When a MapReduce operation is close to completion, the master schedules backup executions of the remaining in-progress tasks. The task is marked as completed whenever either the primary or the backup execution completes.”

§3.6

Vocabulary — as this paper uses it

Map: The user-written function that takes one input key/value pair and produces a set of intermediate key/value pairs, typed map (k1,v1) to list(k2,v2). Its output is buffered in memory and spilled to the worker's local disk.
Reduce: The user-written function that accepts an intermediate key and an iterator over all values for that key, merging them into a possibly smaller set of values. Typically just zero or one output value is produced per invocation.
Master: The one copy of the program that is not a worker: it assigns map and reduce tasks to idle workers, tracks task state, and forwards the locations of intermediate file regions from map side to reduce side. It is also the failure detector and the single point of failure.
Worker: An ordinary copy of the user program that executes whichever map or reduce task the master assigns it. Workers are pinged periodically and their tasks are reassigned if they stop responding.
M and R: The number of input splits, hence map tasks, and the number of output partitions, hence reduce tasks. Both are chosen much larger than the machine count, with M sized so each map task covers 16-64 MB and R a small multiple of the expected worker count.
Partitioning function: The function on the intermediate key that decides which of the R reduce tasks a record belongs to, hash(key) mod R by default. Users can supply their own, for example hashing only the hostname of a URL key so all entries for a host land in one output file.
Combiner: An optional function run on the map machine that partially merges intermediate records with the same key before they are sent over the network. It is applicable when the reduce function is commutative and associative, and is usually implemented by the same code as the reducer.
Straggler: A machine that takes an unusually long time to complete one of the last few map or reduce tasks, whether from a failing disk, contention with other tasks scheduled on it, or a hardware or configuration fault. Stragglers are a leading cause of long job completion times.
Backup task: A redundant execution of a still in-progress task, scheduled by the master when the operation is close to completion. Whichever of the primary or the backup finishes first marks the task complete, at a cost tuned to a few percent of extra resources.

On the timeline — where this sits in the story

Year2004

MilestoneMapReduce

Problem solvedWriting reliable distributed batch jobs over huge datasets required manual scheduling, failure handling, and data distribution.

How it solved itMapReduce let developers write map and reduce functions while the runtime handled distribution, retries, and aggregation.

Why it matteredDefined the first mainstream big-data programming model and inspired Hadoop MapReduce.

DDIACh. 11: Batch Processing

View on the timeline