Combiner function in MapReduce

A combiner is an optional "mini-reducer" that runs on the output of each map task, on the same node, before the data is sent over the network to the reducers. It performs a local aggregation of the map output with the same key, so less data is shuffled.

Purpose

• Reduces network traffic. Map output is written to local disk and transferred to reducers. Combining it first sends fewer records across the cluster, where bandwidth is the scarce resource.
• Reduces disk I/O for spilled map output and the sort/merge work in the reducer.
• Speeds up the job, especially when keys repeat many times in a map's output (e.g. word count).

Properties

• It has the same signature as a reducer: input (key, list of values), output (key, value) with the same types as map output.
• Hadoop may run it zero, one or many times, so the result of the job must not depend on how many times it runs.
• It is valid only for operations that are commutative and associative (sum, count, max, min). Average cannot be used directly (the average of averages is wrong); instead emit (sum, count) pairs.
• It is set in the driver: job.setCombinerClass(MyReducer.class);

Example: word count

Input split across two mappers:

• Mapper 1 text: "big data big"
• Mapper 2 text: "data big data"

Without combiner, mapper 1 sends (big,1) (data,1) (big,1) and mapper 2 sends (data,1) (big,1) (data,1): 6 records.

With combiner, each mapper first sums locally:

• Mapper 1 sends (big,2) (data,1)
• Mapper 2 sends (big,1) (data,2)

Only 4 records cross the network. The reducer then gets big -> [2,1], data -> [1,2] and outputs (big,3), (data,3).

Map1: (big,1)(data,1)(big,1) -> Combiner -> (big,2)(data,1)
Map2: (data,1)(big,1)(data,1)-> Combiner -> (big,1)(data,2)
                    \  shuffle  /
              Reducer: (big,3) (data,3)

Code sketch

public class WordCount {
  public static class TokenMapper
      extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    public void map(LongWritable k, Text v, Context c)
        throws IOException, InterruptedException {
      for (String w : v.toString().split("\\s+")) {
        word.set(w);
        c.write(word, one);
      }
    }
  }
  public static class SumReducer
      extends Reducer<Text, IntWritable, Text, IntWritable> {
    public void reduce(Text k, Iterable<IntWritable> vals, Context c)
        throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable v : vals) sum += v.get();
      c.write(k, new IntWritable(sum));
    }
  }
  // in driver:
  // job.setCombinerClass(SumReducer.class);
  // job.setReducerClass(SumReducer.class);
}

Here the reducer class is reused as the combiner because summing is associative and commutative.