The idea: write two functions, let the framework do the rest
MapReduce (Google, 2004; open-sourced as Hadoop MapReduce) runs one computation over data spread across many machines. The programmer writes only two functions. For counting words, map gets a piece of text and emits the pair (word, 1) for every word; reduce gets one word with all its values, (the, [1, 1, 1, 1, 1]), and emits (the, 5). Everything in between, splitting the input, running tasks on many workers, moving the pairs to the right place, sorting them and surviving crashes, is done by the framework.
The page runs one small job: 13 words in 3 splits, M = 3 map tasks and R = 2 reduce tasks. Time is a tick clock t: a map task takes 3 ticks, the shuffle and sort take one tick, and a reducer handles one input record per tick. The counters at the top of the canvas show how many records the maps wrote and how many records and bytes crossed the network.
| On the page | Meaning |
|---|---|
| split 0, 1, 2 | the input, stored in HDFS blocks |
| master: task table | the state of every task: idle, in progress (on which worker) or completed |
| W1 to W4 | map workers; W4 is an idle spare. The two boxes in each are the map's output on its local disk |
| partition 0 → R0 partition 1 → R1 | one colour per reducer: its partitions, transfers, panel and output file |
| orange boxes | messages from the master: a task, a ping, a kill |
| R0, R1 and part-r-0000x | the reduce tasks (on W5 and W6) and the files they write in HDFS |
The master and its tasks
The master (the JobTracker in Hadoop 1, the MapReduce ApplicationMaster on YARN) cuts the input into splits, usually one per HDFS block, and makes one map task per split. The number of reduce tasks R is set by the user (mapreduce.job.reduces). For every task it keeps a state, idle, in progress or completed, and the worker it runs on. When a worker is free, the master hands it an idle task, preferring a worker that already stores a copy of the task's block, so the input is read from the local disk (Demo 1).
Map and partition
A map task reads its split and calls map() once per word. Each output pair goes into one of R partitions on the worker's local disk, not into HDFS: partition = hash(word) mod R. Hadoop's default HashPartitioner uses the key's hashCode(); the page uses the sum of the letter codes so you can check it by hand: cat = 99 + 97 + 116 = 312, even, so partition 0. Because every map uses the same function, every (cat, 1) in the whole job lands in partition 0 of some map, and all of them reach the same reducer.
Shuffle, sort and reduce
The shuffle moves data from maps to reducers: reducer r fetches partition r from every map, M × R transfers in all (6 here). A reducer cannot call reduce() until every map has finished, because the last map may still hold some of its words. It then merges and sorts what it fetched by key, groups equal keys, and calls reduce(word, [values]) once per group, in key order. Each reducer writes one output file, part-r-00000, part-r-00001: it writes to a temporary file and renames it when it is done, so a half-written file is never visible. In Demo 1 R0 has 5 records and R1 has 8, so the job ends at t = 3 + 1 + 8 = 12.
The combiner: less to shuffle
Split 0 is the cat the dog the: its map sends (the, 1) three times. A combiner is a reduce run inside the map, on its own output, before the shuffle: the three pairs become (the, 3). It only works when reduce is associative and commutative, like a sum or a max; for word count the reducer class itself is used (job.setCombinerClass). The framework may run it zero, one or several times, so the result must not depend on it (Demo 2).
| records shuffled | bytes shuffled | R0 / R1 input | job done | |
|---|---|---|---|---|
| no combiner (Demo 1) | 13 | 74 B | 5 / 8 | t = 12 |
| combiner (Demo 2) | 9 | 52 B | 5 / 4 | t = 9 |
A record is word \t count \n, so the 1 is 6 bytes. On real text, where a few words make up most of it, a combiner often cuts the shuffle by much more than here.
A worker fails: run the task again
The master pings every worker periodically (in Hadoop, tasks report through heartbeats, and a task that is silent for mapreduce.task.timeout, 10 minutes by default, is declared failed). In Demo 3 W2 crashes at t = 2 with M2 two thirds done. At t = 3 the ping gets no reply; the master marks W2 failed, sets M2 back to idle and gives it to W4, which reads split 1 again from another HDFS copy and runs M2 from the start. The job finishes at t = 15 instead of t = 12, with the same output, and nobody had to restart it. Re-running is safe because map and reduce are deterministic functions of their input, and a task's output only counts once the master has recorded it as completed.
A crash also loses completed map tasks of that worker, because their output sits on its local disk, so the master runs those again too. Completed reduce tasks are safe: their output is in HDFS.
Stragglers and backup tasks
A straggler is a worker that is alive but slow: a failing disk, a busy machine. One slow task holds up the whole job. Near the end of a phase, when no task is waiting, the master starts a backup (speculative) attempt of each task still running, and whichever copy finishes first wins; the other is killed (mapreduce.map.speculative, mapreduce.reduce.speculative). In Demo 4 W3 runs at a third of the speed; the backup on W4 finishes at t = 6, where W3 would need t = 9, and the job ends at t = 15 instead of 18. The Google paper reports a sort that took 44% longer with backup tasks turned off.
A skewed key
Hash partitioning spreads keys evenly, not records. In Demo 5, 10 of the 12 words are the, and every one of them goes to R1. R0 is done at t = 6 and sits idle; R1 needs 10 ticks and the job ends at t = 14. Adding reducers does not help, because one key always goes to one reducer. What does help: a combiner (here it would cut the 10 records to 3), splitting a hot key into the#1, the#2 and adding the parts in a second step (salting), or a partitioner that knows the hot keys.
Partitioning by a hash of the key is the same idea as sharding a database by key, with the same hot-spot problem; see Partitioning vs Sharding. Consistent Hashing shows why hash mod N moves almost every key when N changes, which a single MapReduce job never has to worry about because R is fixed for the whole job. On one machine, splitting work into tasks and combining their results is what ForkJoinPool does, with work stealing in place of a master.
Where it is used today
Hadoop MapReduce still runs batch jobs on HDFS, but most new work uses Apache Spark, which keeps data in memory between steps and chains many stages instead of writing to HDFS after every reduce. The ideas on this page are all still there: Spark's stages are separated by the same shuffle (hash-partitioned, written to local disk, fetched by the next stage), reduceByKey combines on the map side, lost partitions are recomputed from their input, and speculation re-runs slow tasks.
What the page leaves out
Real jobs have thousands of tasks, and reducers start copying map output while the last maps run (mapreduce.job.reduce.slowstart.completedmaps). A map sorts and spills its output in a memory buffer before writing partitions, and a reducer merges sorted runs from disk; the page sorts in one step. Input formats, record readers, counters, compression of map output, YARN containers, reduce-task failures and a failing master are left out. Simplifications: one map task per worker, every map task takes 3 ticks, a reducer handles one record per tick, the hash is the sum of the letter codes, and failures are detected exactly one tick later.