MapReduce: the juice factory

From one apple to a container of fruit

The NoSQL lessons spread data and queries over many machines. The first slide of this lecture names what is still missing: "Scalability of computation, not only data and queries" (p. 775). MapReduce is the course's answer, told as a story about juice.

The prototype needs two tools: a knife cuts an apple into slices, a blender turns the slices into juice (p. 778). "Next Day" the recipe meets "all the fruits in the fruit basket" — still one knife, one blender, one glass of mixed juice (p. 779). Then "Industrialization": a giant with a "whole container of fruits" that wants "juice of different fruits separately" (p. 780), one bottle per kind. The slide draws the knife and the blender tiny next to the container: the tools are fine, the scale is the problem. Page 784 names them: the knife is Map, the blender is Reduce.

Read the right panel as the whole model. The container is the input data. The five baskets are the splits, one per map task. Each knife is a mapper: it cuts whatever fruit it gets and leaves every piece tagged with its kind — the key. The criss-crossing lines are the shuffle: every apple piece, from every basket, travels to the one apple pile. Each blender is a reducer: it sees all the pieces of one kind and fills one bottle, an output file.

One machine is too slow, parallel code is hard

Page 781 lists the simple operations — "Making juice", "Indexing", "Recommendations etc" — and gives the verdict on the volume: "Sequential execution just won't scale up". Many machines help, since "Parallel execution achieves greater efficiency", "But, parallel programming is hard" (p. 782): parallelization (with its "Race Conditions" and its "Debugging"), "Fault Tolerance", "Data Distribution", "Load Balancing".

Those are the four hard things of p. 782, absorbed by the framework; an exam answer names both halves of the definition. Page 785 lists them as advantages — "Automatic parallelization, distribution", "I/O scheduling" (load balancing, network and data transfer optimization), "Fault tolerance" ("Handling of machine failures") — and ends with "Need more power: Scale out, not up!": add machines rather than buy a bigger one, the horizontal scaling of lesson 3.

MapReduce is a programming model — you write only map and reduce, over key-value pairs — plus an implementation that runs them on many machines and handles distribution, failures and load balancing.

The word count, by hand and by machine

MapReduce has one data structure: "All data is in terms of key-value pairs" (p. 788). "Mappers read in data from the filesystem, and output (typically) modified data"; "Reducers collect all of the mappers output on the keys, and output (typically) reduced data". Page 789 gives the shape of a "Typical problem solved by MapReduce":

1. "Read a lot of data"
2. "Map: extract something you care about from each record"
3. "Shuffle and Sort"
4. "Reduce: aggregate, summarize, filter, or transform"
5. "Write the results"

You write steps 2 and 4. Step 3 is the library's: it brings all the values of one key together, sorted by key.

The slide's word count

The mapper of p. 792 counts "number of each word in user queries". Its input pair is <queryID, QueryText> — the key Q1, the value the text of the hook — and it emits <word, 1> for every word, punctuation removed: 21 pairs, <The, 1> <teacher, 1> <went, 1> …. A mapper counts nothing: it sees only its own input and cannot know how often a word occurs elsewhere. The table follows the 21 pairs through shuffle and sort to the reducer.

Page 810 writes the same job in pseudo-code, run on a document "Sam Apple Sam Mom Sam Mom":

// emits one <word, "1"> per word
map(String input_key, String input_value):
    // input_key: document name
    // input_value: document contents
    for each word w in input_value:
      EmitIntermediate(w, "1");

// emits the count of one word
reduce(String output_key, Iterator intermediate_values):
    // output_key: a word
    // output_values: a list of counts
    int result = 0;
    for each v in intermediate_values:
      result += ParseInt(v);
    Emit(AsString(result));

The map emits one pair per word — <"Sam", "1">, <"Apple", "1">, <"Sam", "1">, <"Mom", "1">, <"Sam", "1">, <"Mom", "1">; the library groups them into <"Sam", ["1","1","1"]>, <"Apple", ["1"]>, <"Mom", ["1", "1"]>; the three reduce calls emit "3", "1" and "2", each the count of the key of its call. The values travel as strings, hence ParseInt and AsString; Hadoop gives them real types (section 3).

The model in symbols

The "Programming Model" of p. 791: "To generate a set of output key-value pairs from a set of input key-value pairs", \(\{\langle k_i, v_i\rangle\} \to \{\langle k_o, v_o\rangle\}\), "Expressed using two abstractions":

Between the two, the "Library" "aggregates all the all intermediate values associated with the same intermediate key" (sic) and "passes the intermediate key-value pairs to reduce function". One map call may emit no pair, one or many. The slide's reduce emits one pair per key — exact for the word count, though a Hadoop reducer may emit any number.

Two invariants carry the model. "All inputs with the same key must go to the same reducer!" (p. 797): if the four <store, 1> reached two reducers, two files would hold partial counts and no file the total. And "Input is typically sorted, output is output exactly as is" (p. 797): every reducer meets its keys in order, so every output file is sorted — each on its own.

Map tags the pieces of each record with a key, the library gathers all the values of a key in one place, reduce aggregates them: counting happens only after the shuffle.

Seven parts and six types

"MR Key Components" (pp. 794–803) takes the job apart. Fig. 2 puts the parts in the order a pair meets them; the table gives each part in the slides' words.

Two numbers to keep. The slide's 64 MB per chunk ("You can change it", p. 795 adds) is the Hadoop 1 value, equal to the HDFS block size of the time (lesson 31); Hadoop 2 and later use 128 MB. Give 64 MB as the slide's figure. And one split feeds one map task, while the map function runs once per record of the split: once per line in the Java job below.

The partitioner

"Default is simply: Key.hashCode() % numOfReducers" (p. 798); p. 825 writes it as the default partitioning function:

Equal keys have equal hashes, so they always reach the same reducer, while different keys spread roughly evenly. "Custom partitioning is often required": "to produce a total order in the output" — keys a–m to reducer 0 and n–z to reducer 1, so the sorted files concatenate into one sorted list; to "Provide (more) uniform distribution of load between reducers"; because "Some values might need to be sent to the same reducer" — "To compute the relative frequency of a pair of words <W1, W2> you would need to make sure all of word W1 are sent to the same reducer", since count(W1, W2) is divided by the count of all pairs that start with W1; and for "Binning of results". A custom one "Should implement Partitioner interface", "Set by calling conf.setPartitionerClass(MyPart.class)".

The combiner

The combiner runs on the map side, on one mapper's output, before the shuffle: three <store, 1> of one split leave as one <store, 3>. It "Cannot change the type of its input" — "Input types must be the same as output types" (p. 799) — because the reducer must accept combined and raw pairs alike. Page 815 says when it pays — "When maps produce many repeated keys", with the "Goal" "to decrease size of the transient data" — and that "Combiners have the same interface as Reduces, and often are the same class": "In WordCount, conf.setCombinerClass(Reduce.class);". Page 826: a "Mini-reduce phase before the intermediate data is sent to reduce" that "Saves network bandwidth".

Writables: the six types

Keys and values cross machines and are written to disk, so their types "can be serialized / deserialized to a stream": the Writables of p. 803. There are defaults "for basic values, like Strings, Integers, Longs" and for structures such as arrays and maps, and "User can implement this interface". "Your application needs at least six writables": "2 for your input", "2 for your intermediate values (Map <-> Reduce)", "2 for your output" — K1, V1, K2, V2, K3, V3 of Fig. 2. A combiner adds none.

The complete job

Page 811 prints Hadoop's WordCount, inside public class WordCount { … }, in the Java API of the time. Match the type parameters with Fig. 2.

// the Mapper: <offset, line> in, <word, 1> out
public static class Map extends MapReduceBase implements
    Mapper<LongWritable, Text, Text, IntWritable> {
  private final static IntWritable one = new IntWritable(1);
  private Text word = new Text();

  public void map(LongWritable key, Text value,
      OutputCollector<Text, IntWritable> output,
      Reporter reporter) throws IOException {
    String line = value.toString();
    StringTokenizer tokenizer = new StringTokenizer(line);
    while (tokenizer.hasMoreTokens()) {
      word.set(tokenizer.nextToken());
      output.collect(word, one);
}}}

The mapper's input pair is one line of a text file: key its byte offset (LongWritable), value the line (Text). It emits <word, one> with output.collect, the EmitIntermediate of p. 810. StringTokenizer splits at white space only, so "store." and "store" are two different keys for this program.

// the Reducer: <word, [1, 1, …]> in, <word, sum> out
public static class Reduce extends MapReduceBase implements
    Reducer<Text, IntWritable, Text, IntWritable> {
  public void reduce(Text key, Iterator<IntWritable> values,
      OutputCollector<Text, IntWritable> output,
      Reporter reporter) throws IOException {
    int sum = 0;
    while (values.hasNext()) { sum += values.next().get(); }
    output.collect(key, new IntWritable(sum));
}}

// the driver: configures the job and runs it
public static void main(String[] args) throws Exception {
  JobConf conf = new JobConf(WordCount.class);
  conf.setJobName("wordcount");
  conf.setOutputKeyClass(Text.class);
  conf.setOutputValueClass(IntWritable.class);
  conf.setMapperClass(Map.class);
  conf.setCombinerClass(Reduce.class);
  conf.setReducerClass(Reduce.class);
  conf.setInputFormat(TextInputFormat.class);
  conf.setOutputFormat(TextOutputFormat.class);
  FileInputFormat.setInputPaths(conf, new Path(args[0]));
  FileOutputFormat.setOutputPath(conf, new Path(args[1]));

  JobClient.runJob(conf);
}

In main the input format plays the input splitter and the output format the output committer; setCombinerClass(Reduce.class) reuses the reducer as combiner — allowed, because a sum of sums is a sum and the types match. The slide's arrow "Run this program as a MapReduce job" points at JobClient.runJob(conf). Not set: the partitioner, so the default hash is used, and R, so the job runs Hadoop's default of one reduce task.

Splitter, mapper, combiner, partitioner, shuffle and sort, reducer, committer — and a Master that schedules them: you write the mapper and the reducer and choose the rest.

On the cluster: move the code, not the data

"MR: How it works" (p. 804) puts the job on machines, in the figure of pp. 790 and 805: a "Hadoop Program" forks a Master and workers; map workers "read" their splits and "local write" their output, already cut into one region per reducer; reduce workers "remote read, sort" their regions and "write" the output files.

Page 805 pastes a green warning over the "read" arrows: "Transfer peta-scale data through network". If the splits lie on some machines and the map workers run on others, reading moves the whole input across the network. Page 806 answers in six words: "Move the code, not the data". A map function is a few kilobytes; its input is terabytes.

"MapReduce on HDFS" (p. 807) shows how. The Master asks the HDFS NameNode — the server that knows where every block of every file is stored, lesson 31 — "Where are the chunks of input data?", and gives each map task to a worker that stores its split: "Read from local disk". "Disk Locality" (p. 822) adds the fallback: "Map tasks are scheduled close to data", "on nodes that have input data", "if not, on nodes that are nearer to input data" — "Ex. Same network switch" — which "Conserves network bandwidth". The shuffle still crosses the network, since grouping by key mixes every mapper's output; but it carries intermediate pairs, which a combiner shrinks.

How many tasks

"Execution" (p. 808): the "System determines" M, the "no. of map tasks", as "input is partitioned into M splits"; the "User specifies" R, the "no. of reduce tasks"; "intermediate keys are partitioned (using partitioning function) to be processed by desired reduce task". A 1 GB file in 64 MB splits gives M = 16. "Task Granularity" (p. 823) wants "No. of map tasks > no. of worker nodes", for "Better load balancing" — a fast worker simply takes more tasks — and "Better recovery" — a dead worker's small tasks are re-run quickly, on many machines; "But, increases load on Master": "More scheduling decisions", "More states to be saved". "M could be chosen w.r.t to block size of the file system" "to effectively leverage locality"; "R is usually specified by users", and "Each reduce task produces one output file". The figure of p. 809 adds that map worker and reduce worker are roles: the same three nodes run the "Mapping process" and then the "Reducing process".

The execution flow

Page 812 is the figure of the 2004 Google paper by Dean and Ghemawat (the slide does not say so), its steps numbered.

1. Fork. The user program starts copies of itself on the cluster; one becomes the Master, the others workers. Someone must coordinate; everyone else computes.

2. Assign. The Master gives each idle worker a map task or a reduce task. A worker that finishes gets the next task, so fast workers end up doing more.

3. Read. A map worker reads its split — from its own disk when it can — and calls map on each record. The input stays where it is stored.

4. Local write. The map output goes to the worker's local disk, cut into R regions by the partitioner, and the Master records their size and location. The data is temporary: copying it into the distributed file system would cost more than computing it again.

5. Remote read. Each reduce worker fetches its region from every map worker and sorts it by key. This is where the intermediate data crosses the network: grouping by key needs pieces from every map worker.

6. Write. Each reduce worker calls reduce on every key, in order, and writes one output file to the distributed file system, so the result survives any one machine.

The Master's bookkeeping is small (p. 813): "For each task", its "State { idle, in-progress, completed }" and the "Identity of the worker machine"; "For each completed map task", the "Size and location of intermediate data" — the addresses step 5 needs.

Hadoop 1 has its own names for the roles (p. 832, Fig. 5): the JobTracker, one per cluster, is the Master; a TaskTracker on every slave node runs the tasks. The slide never defines them; lesson 31 shows what replaced them.

The Master asks the NameNode where the input blocks are and runs each map task where its split is stored: the code travels, the input stays, and only the shuffle crosses the network.

Dead workers, slow workers, bad records

On a cluster of commodity machines something is always broken. "Fault Tolerance" (p. 821) gives the rules. "Worker failure – handled via re-execution": a dead worker is "Identified by no response to heartbeat messages"; "In-progress and Completed map tasks are re-scheduled"; "Workers executing reduce tasks are notified of re-scheduling"; "Completed reduce tasks are not re-scheduled".

Why re-run map tasks that had finished? Their output lay on the local disk of the dead machine (step 4) and died with it. A finished reduce task wrote its file to the distributed file system (step 6), which keeps copies elsewhere. A reduce task the dead worker was still running starts again elsewhere too: the slide lists only the map side, but in the Google paper every in-progress task of a failed worker goes back to idle. Re-running is safe only if a task gives the same output twice: "our nodes are (or at least should be) deterministic", "Nodes should have no side effects!" (p. 802) — node means a task there. A task that also inserted rows into an outside database would insert them twice. A silent task is treated the same way: "If a task fails to report progress … it is assumed to be stuck, and is killed, and the step is re-launched (with the same input)" (p. 801).

"Master failure" is "Rare"; it "Can be recovered from checkpoints", or "All tasks abort" (p. 821): two designs. A Master that saves checkpoints of its tables restarts from the last one; one that saves none aborts the job, and the client starts it again.

Stragglers and back-up tasks

A job ends when its last task ends. "Slow workers delay completion time" (p. 824), for reasons as dull as "Bad disks with soft errors", "Other tasks eating up resources" or "Strange reasons like processor cache being disabled". The remedy: "Start back-up tasks as the job nears completion"; "First task to complete is considered". Hadoop calls it speculative execution (p. 818): "The framework can run multiple instances of slow tasks", "Output from instance that finishes first is used", switched by mapred.speculative.execution; it "Can dramatically bring in long tails on jobs". The losing copy is killed — "First one to complete wins, then any other runs are killed" (p. 802) — and the committer lets only the winner's file through.

Back-ups cure slow machines, not heavy tasks. A reducer that receives far more values than the others is slow on any machine, and so is its copy. One cause is the partitioning function: "hash(key) mod R" by default, it must be "well-balanced" — "If not, reduce tasks may delay completion time" (p. 825) — because it may pile several big keys on one reducer. Another is one huge key, which no partitioning function can split, since all its values must reach the same reducer; there a combiner, when the reduce function allows one, shrinks the key on the map side.

Skipping bad records

Some crashes repeat: "Map/Reduce tasks may fail on certain records due to bugs" (p. 827). "Ideally, debug and fix" — "Not possible if third-party code is buggy". So "When worker dies, Master is notified of the record", and "If more than one worker dies on the same record", the "Master re-schedules the task and asks to skip the record". One record lost out of billions is fine for statistics, not for a bank balance.

Failures are handled by re-running deterministic tasks — finished map tasks too, since their output was on the dead machine's disk — and slow tasks get a back-up copy near the end: the first copy to finish wins.

Tools, schedulers, and the end of disk locality

Page 814 lists "Some further handy tools": "Combiners", "Compression", "Counters", "Speculation". Two you have met. Compression (p. 816): "Compressing the outputs and intermediate data will often yield huge performance gains", and it "Can be specified via a configuration file or set programmatically". Compressing the map output shrinks exactly what the shuffle moves.

Counters (p. 817): "Often Map/Reduce applications have countable events"; the "framework counts records in to and out of Mapper and Reducer", and you can add your own — malformed lines, skipped records — through the Reporter that map and reduce receive, with "nice names in a MyClass_Counter.properties file". Each task counts locally; the framework adds up the totals.

// declare two user counters, then add 1 to the first
static enum Counter {EVENT1, EVENT2};
reporter.incrCounter(Counter.EVENT1, 1);

Zero reduces (p. 820): "Frequently, we only need to run a filter on the input data", with "No sorting or shuffling required by the job". "Set the number of reduces to 0", and the "Output from maps will go directly to OutputFormat and disk": one file per map task, no shuffle at all.

Whose job runs first

A shared cluster runs many jobs, and something decides whose tasks get the free slots. "By default, Hadoop uses FIFO to schedule jobs" (p. 829): first submitted, first served, so a long job holds up every short one behind it. "Alternate scheduler options: capacity and fair".

Capacity gives each organisation a slice of the cluster and lends it while idle, but a borrowed slot is not taken back while its task runs. Fair gives every running job its share over time, so a small job submitted behind a big one gets slots as soon as running tasks free them, instead of waiting for the big job to end.

The end of disk locality

"New Trend: Disk-locality Irrelevant" (p. 828). Moving the code to the disk "Assumes disk bandwidth exceeds network bandwidth". But "Network speeds fast improving", "Disk speeds have stagnated": a remote disk now costs about as much as a local one. "Next step: attain memory-locality" — keep the data in RAM between steps, the idea of Spark in lesson 38. The "Conclusion" (p. 833) sums up: an "Easy to use scalable programming model for large-scale data processing on clusters" that "Hides issues of parallelization, fault tolerance, data partitioning & load balancing" and "Achieves efficiency through disk-locality".

Back to <store, 3>

Could a cluster turn the four occurrences of store into 3? Not by losing a pair: a dead mapper's pairs are produced again by re-execution. Not with a combiner: <store, 2> and two <store, 1> still add up to 4. Not with two reducers: the partitioner sends every store pair to the same one. The framework only moves and groups what the mappers emit; its one exception, a bad record skipped after it crashed more than one worker (p. 827), drops a whole record, not one word. So the count can change only in your own code — and the Java mapper of p. 811 does change it: StringTokenizer keeps the full stop, and the job writes <store, 3> next to <store., 1>, as the first widget shows. The slide's own mapper output has four <store, 1>, so its 3 is a slip of the pen.