The previous lesson ended with a forty-line coordinator that split chunks of orders.jsonl among processes, added up partials and requeued a dead worker's task. It worked on a laptop; on a thousand-node cluster you would have to solve, for every job, data locality, the shuffle between nodes, failure detection, speculative execution and atomic output. MapReduce is the programming model with which Google (Dean and Ghemawat, 2004) solved all of that once and for all, so that the programmer writes just two functions, map and reduce, and the system takes care of the rest. Hadoop is its open-source implementation, together with HDFS (04-02) and the YARN resource manager, and for a decade it was synonymous with "big data". Hardly anyone writes MapReduce jobs by hand any more, but everything that came afterwards (Spark, Flink, distributed SQL engines) uses its vocabulary and its phases, and its design decisions (always write to disk, independent tasks, a master that reassigns) are still the baseline against which improvements are explained. This lesson presents the model using Kilometre Zero's sales by producer and by market, walks through the execution of a job inside YARN, and implements it three times: in Python with Hadoop Streaming, in Java as the canonical example, and as a local simulation of the shuffle so you can get your hands on what the framework hides.

1. The programming model: map, shuffle & sort, reduce
2. Combiner and partitioner
3. Execution flow and fault tolerance
4. Hadoop: HDFS, YARN and MapReduce v2
5. Anatomy of a job: from submit to _SUCCESS
6. Input and output formats
7. Why MapReduce is slow and where it stands today
8. Hands-on: sales by producer and market in Streaming, in Java and in a simulation
9. Common Mistakes and Tips
10. Exercises
11. Conclusion

MapReduce borrows two names from functional programming and gives them a precise meaning for distributed data. Every job processes key/value pairs and goes through three phases:

1. Map. The system divides the input into chunks (splits) and, for every record in every chunk, runs the function map(k1, v1) → list of (k2, v2). The programmer decides what the intermediate key k2 is: it is the key they want to group by. For "sales by producer", map receives a line of orders.jsonl and emits, for each line of the order, (producer, quantity × price).
2. Shuffle & sort. The system collects all the (k2, v2) pairs from all the mappers, sends them to the reducer responsible for each k2, and delivers them grouped by key and sorted: (k2, [v2, v2, v2, ...]). This is the phase the programmer does not write and the one that costs the most (05-01, section 5).
3. Reduce. For each key, reduce(k2, list of v2) → list of (k3, v3). For sales, it adds up the list and emits (producer, total).

The example MapReduce is always introduced with is the word count: map emits (word, 1) for every word in a line and reduce adds up the ones. It is the same skeleton as ours with "word" swapped for "producer" and "1" for "amount", which is why MapReduce is said to be a generalised word count: any computation that can be expressed as "extract a key from each record and aggregate by key" fits directly; those that need several chained groupings (sales by producer and then the top-selling producer per market) are expressed as several jobs in a chain, each one reading the previous one's output from HDFS, which is the source of the slowness in section 7.

Three properties of the model explain its success:

• The functions are local. map sees one record; reduce sees one key and its values. Neither needs to know how many nodes there are or where the data is. The programmer writes business logic, not distribution.
• The tasks are independent. Each map over a split and each reduce over a partition of keys can run on any node, in any order, and be repeated: this is the deterministic re-execution of 05-01.
• The shuffle is generic. A single mechanism (partition by key, sort, transfer, merge) serves every job, and the system can optimise it for everyone.

The basic model has two hooks the programmer can replace:

Combiner. A mapper processing a 128 MB block of orders.jsonl emits about 400,000 pairs, but there are only three distinct producers: 400,000 pairs will travel over the network so that the reducer can add up 133,000 values per key. The combiner is a reduce function local to the mapper that runs over each map's output before the shuffle: it groups the mapper's pairs by key and reduces them to one per key. With it, the mapper sends 3 pairs instead of 400,000. This is the "reduce before you move" of 05-01. It is only valid when the reduction is associative and commutative, and has the same input and output type: a sum qualifies; an average does not, unless you carry (sum, count). Hadoop does not guarantee that the combiner will run (it may run zero, one or several times over the same data), so the result must be identical with or without it.

Partitioner. It decides which reducer each key goes to: partition(k2) = hash(k2) mod number_of_reducers by default. You replace it when you want to control the distribution: sending all the keys in a range to the same reducer (to obtain globally sorted output), or separating two kinds of key into two output files. In the hands-on part of section 8 we emit two families of keys in the same job (p:<producer> and m:<market>) and a partitioner sends them to different reducers, so that part-r-00000 holds sales by producer and part-r-00001 sales by market. The partitioner is also the place where skew is tackled: a partitioner that knows the hot keys can spread them over several reducers, with a second pass to combine.

The original paper describes an architecture with one master and many workers, which Hadoop keeps under other names (section 4):

1. The client submits the job: the code (a jar or scripts), the configuration and the input and output paths.
2. The master asks HDFS for the input's blocks and creates one map task per split, noting which nodes each block lives on. It also creates R reduce tasks, with R configured by the user.
3. It assigns map tasks to free workers, preferring the one that has the block on its disk; failing that, one in the same rack; failing that, any.
4. Each map writes its output, partitioned and sorted, to the worker's local disk, and tells the master where it is.
5. When all the maps have finished, the master assigns the reduce tasks; each reducer fetches its partition of every map's output over the network, merges it while preserving the order, and runs reduce key by key.
6. Each reducer writes its output file to HDFS. When they have all finished, the job is complete.

Fault tolerance rests on what we already know:

• Worker failure. The master detects it through missed heartbeats. The map tasks completed on that node are re-run even though they had finished, because their output was on the local disk of the dead node and the reducers that had not yet fetched it need it. Completed reduce tasks are not re-run: their output is in HDFS. Tasks in progress go back to the queue. Exactly what work_queue.py did, with the added subtlety of the local intermediate output.
• Task failure (an exception in the code, a corrupt record). It is retried up to four times (mapreduce.map.maxattempts); if it keeps failing, the job fails (or, if so configured, a percentage of failed tasks is tolerated in order to skip poison records, the batch version of the DLQ from 02-05).
• Stragglers. Speculative execution (05-01): when the phase is close to finishing, a copy of the slow tasks is launched.
• Atomic output. Each task writes to a temporary directory (_temporary/attempt_.../) and only when the task finishes does the framework do the commit: it renames the task's file to part-r-00001 in the output directory. If two attempts of the same task (re-execution, speculation) finish, only the first one commits. When the job ends, an empty _SUCCESS file is created as a signal to whoever consumes the output (the sensor in 05-05 will wait for it). It is the os.replace of 05-01, institutionalised.
• The master as a single point. In the original design, if the master went down, the whole job was aborted and the client relaunched it: this was accepted as a reasonable trade-off because a master is one machine among thousands and a job can be relaunched. Hadoop 2 improved on it by creating one master per job (the ApplicationMaster of the next section) that YARN can restart, and the ResourceManager has high availability with ZooKeeper (03-03).

Since version 2, Hadoop has been three stacked projects:

In Hadoop 1, the MapReduce master (the JobTracker) did two things at once: managing the cluster's resources and coordinating every job. That made it a bottleneck (a limit of about 4,000 nodes) and tied the cluster to MapReduce: nothing else could run on it. YARN separated the two roles:

• ResourceManager (RM). One per cluster (with high availability). It knows each node's resources and arbitrates between applications with a scheduler (the Capacity or Fair Scheduler, with per-team queues: analytics has its own queue with 40% of the cluster guaranteed). It knows nothing about maps or reduces.
• NodeManager (NM). One per node. It reports its available CPU and memory to the RM, launches and supervises containers (a process with a CPU and memory quota, implemented today with cgroups) and serves the maps' intermediate output to the reducers (the shuffle service).
• ApplicationMaster (AM). One per application (per MapReduce job), running in an ordinary container. It is the master of section 3: it negotiates containers with the RM, asks the NMs to launch tasks in them, tracks their progress and re-runs the failed ones. If the AM dies, the RM restarts it (up to yarn.resourcemanager.am.max-attempts times) and the new AM recovers the progress from the log of completed tasks.
• Container. The unit of allocation: "1 core and 2 GB on node dn-07". Each map or reduce task runs as a JVM inside a container.

The architectural consequence of YARN is that the cluster is a shared resource where different applications live side by side (a MapReduce job from analytics, a Spark job from the recommendations team, a long-running service), and that the compute model is interchangeable: when we launch Spark on YARN in 05-03, the Spark driver will be the ApplicationMaster and the executors will run in containers, with the same RM arbitrating. Kubernetes plays that same role of generic resource manager today (07-05).

It is worth following closely what happens inside a task, because the names reappear in Hadoop's web UI, in the job counters and in explanations of why a job is slow.

Map side.

1. The InputFormat (section 6) computes the splits: by default, one per HDFS block, adjusted to the end of a line just as chunks_by_bytes did in 05-01. With a 150 MB orders.jsonl, that is two splits, two maps.
2. The RecordReader hands records to map: for text, (byte offset, line).
3. map emits pairs into a circular in-memory buffer (100 MB by default, mapreduce.task.io.sort.mb). When it is 80% full, a thread spills it to disk: it partitions by reducer, sorts by key within each partition, applies the combiner if there is one, and writes a spill file.
4. When the map finishes, the spill files are merged into a single one, partitioned and sorted, with an index that says where each partition begins. If there were several spills, the combiner is applied again during the merge.

Reduce side.

5. Copy (fetch). As soon as a map finishes, each reducer asks the shuffle service of that map's NodeManager for its partition, over HTTP, with several threads in parallel. It does not wait for all the maps to finish before it starts copying (but it does before it starts reducing).
6. Merge and sort. The fragments received, each one already sorted, are merged (in memory if they fit, on disk in rounds if they do not) into a single sequence sorted by key.
7. Reduce. The sequence is scanned; every time the key changes, reduce(key, iterator of values) is called. That is why the reducer receives an iterator, not a list: the values for one key may not fit in memory (Montblanc Dairy's 133,000 amounts without a combiner).
8. Commit. The output goes to _temporary/, and on completion it is renamed to part-r-0000N. When the AM confirms that all the reduces have committed, it writes _SUCCESS.

The job counters summarise all of this. For a real job over one campaign day (250,000 events, 150 MB, 2 maps, 2 reduces), with and without a combiner:

Two things to read from this: the shuffle bytes drop by four orders of magnitude with the combiner, and even so the job takes 52 s for what sales_scatter_gather.py did in 1 s. That minute is the framework's fixed cost: starting the AM, requesting containers, launching four JVMs, writing spills and merging, committing to HDFS. A MapReduce job makes no sense below the gigabyte range, and that is the point Spark goes after.

The InputFormat decides two things: how the input is divided into splits and how the records in each split are read. The OutputFormat decides how the results are written.

Two practical warnings. First: a format is splittable only if you can start reading from the middle, and that also depends on the compression: gzip is not splittable (a 1 GB gzip file is a single split and a single map, however big it is), whereas bzip2, indexed LZO and the container formats (Avro, Parquet, ORC, SequenceFile) are. Second: HDFS and MapReduce suffer with small files (04-02): 10,000 files of 50 KB are 10,000 maps lasting an instant each, and the scheduling cost dominates; they have to be consolidated (CombineFileInputFormat, or better, a prior compaction step).

The output follows the same logic: TextOutputFormat writes key<TAB>value per line into one part-r-NNNNN per reducer; with LazyOutputFormat no empty files are created; MultipleOutputs lets a reducer write to several named files (sales by producer in producers-r-00000, by market in markets-r-00000). The number of output files is always the number of reducers, and that is the reason to choose it carefully: few reducers make for large files and long tasks; many, for small files that will be a problem for the next job.

The decisions that made MapReduce robust are the ones that make it slow:

• Everything goes through disk. Each map's output is written to local disk; the reducer copies it and writes it to disk again when merging; the reduce output goes to HDFS with three replicas. A job with one shuffle means at least four writes of the intermediate data volume. It was done this way so that any task could be re-run by reading its input from disk without depending on the memory of a node that may die.
• Jobs are chained through HDFS. A computation with several groupings (sales by producer, then a ranking per market, then a join with the catalogue) is three jobs, and between one and the next the complete output goes to HDFS with replication and the next job reads it back. Iterative algorithms (recommendations by gradient descent, PageRank, the BSP of 05-01) are ten or a hundred chained jobs, each one re-reading the entire dataset.
• Fixed cost per job and per task. Starting a JVM per task, negotiating containers, the commit. Tens of seconds that do not matter in a three-hour job and are everything in an interactive query.
• Rigid model. Only map and reduce; a join between two datasets has to be expressed by emitting both sides with the same key and telling them apart in the reducer (a reduce-side join), or by loading the small one into each mapper's memory (a map-side join, the forerunner of the broadcast join in 05-03). The framework does none of that for you.

Its place today is that of a historical foundation and mental model: the vocabulary (map, shuffle, reduce, combiner, partitioner, split, counters) is the one Spark and Flink use; YARN and HDFS are still in production at many companies as the resource and storage layers, with Spark on top; and Hive, the SQL data warehouse on Hadoop that Facebook created to avoid writing MapReduce by hand, still exists but runs its queries on Tez or Spark rather than on MapReduce, keeping its table catalogue (the metastore) as a central piece of many data lakes. Seeing a new MapReduce job in 2026 is rare; understanding how it worked is what lets you read a Spark plan or the Flink dashboard without surprises.

The job is the same in all three versions: read /km0/events/2026-09-14/orders.jsonl (the file that upload_events_hdfs.py left in HDFS in 04-02, or the one generated by sales_scatter_gather.py in 05-01), and produce total sales by producer and by market. To do it in a single job, the mapper emits two keys per order line, with a prefix: p:montblanc-dairy and m:girona. A partitioner sends the p: keys to reducer 0 and the m: keys to reducer 1, so the output ends up in two clean files.

Hadoop Streaming lets you write the mapper and the reducer in any language that reads from standard input and writes to standard output. Hadoop launches the process, feeds it the records line by line and collects what it emits; the key and the value are separated by a tab.

The Streaming reducer does not receive (key, list of values) as in Java, but the sorted sequence of pairs: it is the script itself that detects the change of key. That is the reason the shuffle sorts rather than just groups: with sorted input, grouping means comparing with the previous line, with no memory needed. And since the reducer adds up, it also works as a combiner unchanged.

Before touching the cluster, a Unix pipeline reproduces the whole job, with sort playing the part of the shuffle:

(Over the three sample lines from 05-01: Anna in Girona bought 2 × 3.90 + 12.50 = 20.30; Mark in Lleida 6 × 9.80 = 58.80; Lucy in Valencia 3 × 4.20 + 4 × 1.60 = 19.00; and by producer, La Vega Farm 7.80 + 6.40 = 14.20, Montblanc Dairy 12.50 + 12.60 = 25.10, Roble Alto Winery 58.80.) This local test is worth its weight in gold: most errors in a Streaming job (a split that fails, a non-numeric value) show up here in a second rather than in a minute of failed job.

Launching it on the cluster from the docker-compose.yml of 04-02 (with YARN added: a resourcemanager and a nodemanager from the apache/hadoop image):

Line by line: -files copies the scripts to every container (the distributed cache: the code travels to the data); -combiner reuses the reducer locally in each map; -partitioner with KeyFieldBasedPartitioner and the option -k1.1,1.1 partitions by the first character of the key (p or m), so that with two reducers each family goes to one (with the hash of p and m modulo 2 they land on different reducers; if they did not, a custom partitioner would do the job, which in Streaming can only be written in Java). The output directory must not exist; MapReduce refuses to overwrite, precisely to protect the atomic output. The result:

And the state of the job, with its counters, can be checked with yarn application -list -appStates ALL, mapred job -status <job_id> or in the ResourceManager's web UI (http://localhost:8088).

Java is Hadoop's native language, and a job in Java avoids the cost of launching an interpreter per task and of serialising everything as text. This is the canonical example, line by line:

What you need to understand about each block:

• The Writable types. Hadoop does not use String or double in its interfaces, but Text, DoubleWritable, LongWritable: types that serialise compactly and compare byte by byte, which the shuffle can sort without deserialising. Reusing the instances (key.set(...) instead of new Text(...)) is the most frequently cited MapReduce optimisation: a mapper emits millions of pairs and creating objects would send garbage collection through the roof.
• Mapper<K1, V1, K2, V2> and Reducer<K2, V2, K3, V3>. The generics document the contract of section 1. Context is the channel through which the task emits pairs (ctx.write) and increments counters.
• The reducer's Iterable can be traversed only once: Hadoop feeds it from the sorted sequence on disk, and reuses the DoubleWritable object on every iteration (keeping references to the values is a classic mistake: they all point to the same object).
• Job is the declarative description: which classes, how many reducers, which types, which paths. waitForCompletion submits it to the ResourceManager and blocks until it finishes. setJarByClass tells Hadoop which jar contains the code, which YARN will copy to every container.
• setCombinerClass(SumReducer.class) is valid because the reducer's input and output are of the same type (Text, DoubleWritable) and addition is associative. If the reducer emitted something else (a ranking, an average) you would need a different combiner or none at all.

Building and launching:

The output is identical to the Streaming one (with part-r-00000 and part-r-00001, the r indicating that a reducer wrote it) and it takes a few seconds less thanks to the absence of Python interpreters and intermediate text. In a job over terabytes, that difference is tens of minutes.

To see what the framework hides, simulations/mapreduce_local.py implements the three phases in a single process, with the shuffle made explicit: partition, sort by key, group.

With the 400,000-order file from 05-01 and --no-combiner, each reducer receives hundreds of thousands of pairs; with the combiner, it receives 3 × n_maps for the producers and 4 × n_maps for the markets. The groupby from itertools over the sorted sequence is literally what the Streaming reducer does when it detects the change of key, and sorted over the fragments is the merge of section 5 (in Hadoop, a merge of already-sorted sequences, cheaper than a full sort).

• Testing directly on the cluster. A Streaming job is tested with cat | mapper | sort | reducer in a second; on the cluster, every failed attempt costs a minute and a YARN log you have to go and dig out with yarn logs -applicationId.
• A combiner that is not associative. Computing averages, or "the first one", or emitting a different type in the combiner, produces results that change depending on how many times it has run. Rule: the combiner must be a function such that running it 0, 1 or N times gives the same final result.
• Keeping references to the values of the Iterable. Hadoop reuses the object; if the reducer does list.add(v) in order to sort later, the list ends up with N copies of the last value. You have to copy (new DoubleWritable(v.get())).
• Existing output directory. The job fails before it starts. This is intentional: atomic output demands a clean directory. Deleting it is part of the pipeline (05-05), not of the job.
• gzip on the input. A 2 GB orders.jsonl.gz is one split and one twenty-minute map. Use bzip2, indexed LZO, or better still Parquet.
• Too many or too few reducers. A single one makes the reduce sequential (Amdahl); a thousand over 20 MB of data create a thousand tiny files. As a guide: have each reducer process between 1 and 5 GB of shuffle, and never more reducers than useful distinct keys.
• Ignoring the counters. Reduce shuffle bytes and Spilled records are the job's thermometer. If the shuffle is the size of the input, the combiner is missing; if the spills are several times the map output, the sort buffer is short of memory.
• Mixing job chaining with logic. Three jobs chained by hand with intermediate paths in HDFS turn into a fragile pipeline. That is the job of the scheduler in 05-05, and a compelling reason to move to Spark, where the three phases are a single program.

Design a MapReduce job (or chain of jobs) that produces, for each market, the producer with the highest sales and its amount. State the intermediate keys and values of each phase, whether you can use a combiner, and how many jobs are needed. Then write the Streaming mapper.py and reducer.py for the first job.

The job in section 8.1 has 2 maps and 2 reduces. Describe what the ApplicationMaster does in each of these cases and which tasks are re-run: (a) the NodeManager dn-07 dies when map 1, which was running there, had already finished and reduce 0 had copied its partition but reduce 1 had not yet done so; (b) reduce 1 throws an exception because of a non-numeric value on its third line; (c) reduce 0 takes five times the median and the AM launches a speculative copy that finishes first. Which files are there in /km0/aggregates/2026-09-14/sales/ during and after each case?

During Artisan Cheese Week, p:montblanc-dairy accounts for 50% of the pairs. With the combiner enabled, is that a problem? And without the combiner, or if the reduce were "list of the 100 largest orders for each producer" (where the combiner does not cut the volume as much)? Propose a Java partitioner and the logic of a second job to spread the hot key over four reducers and combine afterwards, following the salting of 05-01.

Exercise 1.

Two jobs are needed, because there are two chained groupings: first add up by (market, producer) and then, per market, pick the maximum.

• Job 1. map: for each order line, (market|producer, amount). Combiner: sum (associative). reduce: sum. Output: girona|montblanc-dairy<TAB>18240.50.
• Job 2. map: reads the output of job 1 and emits (market, producer|total). Combiner: yes, "keep the maximum" is associative and commutative and does not change the type. reduce: go through each market's values and emit the producer with the highest total.

Job 1 in Streaming:

A single job would be possible with a composite key market and values producer|amount, adding up by producer inside the reducer with an in-memory dictionary: it works if one market's producers fit in memory (here they do, there are three), but it loses the combiner and loads the whole volume onto the reduce. With Spark it will be a groupBy followed by a window, in a single program (05-03).

Exercise 2.

(a) The AM stops receiving heartbeats from dn-07 and marks its tasks as lost. Map 1 was completed, but its output lived on the local disk of dn-07, and reduce 1 had not yet copied it, so the AM re-runs map 1 on another node (requesting a container from the RM, preferably on a node with a replica of the block). Reduce 0 keeps its copy and is not affected; reduce 1 waits and copies from the new location. During that time, the output directory contains only _temporary/ with the attempts in progress.

(b) The reduce 1 task fails with an exception; the AM retries it in another container (up to 4 attempts by default). Since the error is deterministic (the same corrupt value), it will fail all four times and the whole job fails; _SUCCESS is not written, and _temporary/ is cleaned up. Solution: have the reducer tolerate the value (try/except with a km0,non_numeric_values counter) or configure mapreduce.reduce.failures.maxpercent. Reduce 0, which had finished and committed, leaves its part-00000 in the directory, but without _SUCCESS no consumer should read it.

(c) The two copies of reduce 0 write to _temporary/attempt_..._r_000000_0/ and _temporary/attempt_..._r_000000_1/. The speculative copy finishes first and requests the commit; the AM grants it and renames its file to part-00000; it then kills the original attempt and discards its temporary directory. In the end: _SUCCESS, part-00000 (from attempt 1) and part-00001. There are never two part-00000 files because the commit is exclusive per task.

Exercise 3.

With a combiner and a sum, it is not a problem: each mapper reduces its 200,000 Montblanc pairs to one, and the reducer receives one per mapper. The skew is in the mappers' input, which is already split by blocks, not by key. Without a combiner, reducer 0 receives half of the job's pairs and takes twice as long as the markets reducer; with a "top 100 per producer", the combiner only reduces each mapper to 100 pairs per producer, which is still not much, so that is not serious either; skew really matters when the reducer needs all the values (the complete list of a producer's orders, an exact median).

Salting partitioner:

The mapper adds the suffix #(hash(order_id) % 4) only to the hot key (deterministic: re-runnable). Job 1 produces four partials p:montblanc-dairy#0..3; a job 2 (or a lightweight final step outside MapReduce, because it is four numbers) strips the suffix and adds up. For the "top 100", job 2 merges four lists of 100 and keeps the best 100: correct because the global top-K is contained in the union of the partial top-Ks.

MapReduce turned the patterns of 05-01 into a two-function contract: the programmer writes map, which extracts a key from each record, and reduce, which aggregates the values of a key, and the system supplies locality (one map per block, run where the block lives), the shuffle & sort that groups by key, the re-execution of failed tasks, speculative execution and atomic output with _temporary and _SUCCESS. The combiner is the pre-reduction that saves four orders of magnitude of shuffle in the sales job, and the partitioner is the lever for separating families of keys or breaking up a hot key. Hadoop implements it on top of HDFS with YARN as a generic resource manager, with a ResourceManager that arbitrates, NodeManagers that launch containers and one ApplicationMaster per job acting as a restartable master. We have run it three times: in Python with Hadoop Streaming, tested first with cat | mapper | sort | reducer; in Java, the canonical example with its Writable types, its Job and its single-use iterator; and in a simulation that makes the partitions, the sort and the groupby visible.

We have also seen the price: every phase writes to disk, every additional grouping is another job that goes through HDFS, every task starts a JVM, and a computation that a Python process solves in a second takes a minute on the cluster. For a nightly batch of terabytes that is an acceptable price; for chaining the sum by producer with the catalogue join and the ranking per market, or for training the recommendations in a hundred iterations over the clicks, it is not. The next lesson introduces Spark, which keeps the model (partitions, shuffle, re-runnable tasks) but expresses it as a DAG of operators that runs in memory, with an optimiser that decides the stages: the daily_sales.py of analytics will go from two scripts and a hadoop jar to a thirty-line program with DataFrames.