Combiner

1 The story

Two crore ones on the network

Kabir runs word count on the 102 MB feedback file and opens the counters. “Reduce shuffle bytes = 225,867,818.” The map tasks emitted 2 crore 6 lakh pairs, and nearly all of them were (“was”, 1), (“the”, 1), (“chai”, 1): the same few words, each with a 1, copied one by one to a single reducer. Riya adds one line to the driver and runs it again. Same answer, 29 seconds instead of 53, and the shuffle shrinks to 1,083 bytes.

Riya

One line, and the shuffle went from 226 MB to one kilobyte? What did it do?

Arjun Sir

job.setCombinerClass(IntSumReducer.class). It runs your reducer's logic on each map task's output before the shuffle. Instead of sending “chai, 1” a lakh times, every map sends “chai, its total”. The reducer then adds three totals instead of lakhs of ones.

Kabir

Then let's use it everywhere. My average-spend job could use the same trick.

Arjun Sir

Try it. Your averages will come out wrong for all 2,000 students, and Hadoop won't say a word. Today we see why a combiner may run zero, one or several times, which kinds of logic survive that, and how to fix the average with a (sum, count) pair.

2 Definition

Formal definition (write this in exams)

A combiner is an optional, user-supplied function in a MapReduce job that performs local aggregation of the intermediate key-value pairs produced by a map task, before they are transferred to the reducers. It is set with job.setCombinerClass() and is implemented as a Reducer whose input and output key and value types are both equal to the map output types. The framework applies it to sorted map output when spilling the sort buffer and may apply it again when merging spill files (when there are at least mapreduce.map.combine.minspills = 3 spills), and on the reduce side during merges; it may therefore be executed zero, one or many times on any subset of a key's values. Consequently the combining operation must be commutative and associative and must not change the final result, regardless of how often it runs. Its purpose is to reduce the volume of data written to local disk and transferred in the shuffle.

In simple words

A combiner is a mini-reducer that each map task runs on its own output, so it sends “chai: 1,33,000” instead of 1,33,000 lines of “chai: 1”. Hadoop decides when, and how many times, to run it, so it is only safe for logic like adding, counting, max and min.

Words you'll meet in this lesson

CombinerLocal aggregation of one map task's output before the shuffle.

setCombinerClassThe driver line that turns it on; often the same class as the reducer.

CommutativeOrder doesn't matter: a + b = b + a.

AssociativeGrouping doesn't matter: (a + b) + c = a + (b + c).

Combine input / output recordsCounters showing how much the combiner shrank the data.

minspillsmapreduce.map.combine.minspills (3): from this many spills, the combiner also runs while merging them.

Partial aggregateA value that can be merged again later, e.g. (sum, count) instead of an average.

In-mapper combiningAggregating in a HashMap inside the mapper and writing totals in cleanup().

From our lab (Hadoop 3.5.0, Module 8 lab cluster)

Word count, no combiner	102 MB, 3 maps: 20,664,760 pairs → 225,867,818 shuffle bytes, Spilled Records 61,994,280, 52.9 s
Word count, combiner	Combine input 20,665,132 → output 465; Reduce input records 93; shuffle 1,083 bytes; Spilled Records 558; 29.2 s; output identical
Average, wrong combiner	The averaging reducer used as a combiner: 2,000 of 2,000 students wrong, e.g. S1413 25.30 instead of 25.13, “3 orders” instead of 184
Average, right combiner	Combiner adds (sum, count) pairs; shuffle 6,500,018 → 107,140 bytes; 0 of 2,000 differ

The combiner at three levels

Pick your level. New to this? Start at Beginner and climb.

Beginner Tally sheets at counting tables

Without a combiner, every counting table sends a slip for each single vote to the central desk: lakhs of slips saying “Party X: 1”.
With a combiner, each table first writes one tally sheet: “Party X: 4,312”. The central desk adds a handful of sheets.
Adding is safe to do in pieces: the total is the same however the votes were grouped. That is the whole rule.

Intermediate Where it runs and what the counters say

The combiner runs inside the map task, on sorted map output, each time the sort buffer spills (Lesson 8.4).
With 3 or more spills it runs again while those spills are merged. That is why Combine input records (20,665,132) was larger than Map output records (20,664,760): 372 records produced by earlier combines were combined a second time.
The arithmetic: 12 spills × 31 words = 372 records, merged and combined into 3 maps × 31 words = 93 records. Combine output 372 + 93 = 465. The reducer received exactly 93.
Its input and output types must both equal the map output types, because its output is fed to the reducer as if the map had written it, and may even be fed to the combiner again.

Advanced What is safe to combine

Safe: sum, count, max, min, logical AND/OR, set union. Applying them to partial results gives the same final answer.
Not safe as-is: average, median, percentiles, “count distinct” by counting, “top-N” on partial counts. An average of partial averages ignores how many values each one stood for.
The fix: make the map emit a partial aggregate that can be merged again: (sum, count) for average, (min, max), a set of IDs for distinct. Divide or finish only in the reducer.
Hadoop gives no guarantee that the combiner runs at all, so the reducer must produce the right answer whether or not it ran. Never put side effects (writing files, external calls) in a combiner.

Exam tip: For “What is a combiner? When can't you use one?”, write: local mini-reducer on map output, set with setCombinerClass, reduces shuffle data; may run 0, 1 or many times, so the function must be commutative and associative with input type = output type = map output type. Example that works: word count (sum). Example that fails: average (use sum and count instead).

3 Types and ways to use it

Which jobs can use a combiner?

Job	Combiner	Why	What the map should emit
Word count, orders per item	The reducer itself	Sum is commutative and associative	(key, count)
Max / min order value	The reducer itself	max(max(a, b), c) = max(a, b, c)	(key, value)
Average per student	A separate (sum, count) combiner	Averages of averages are wrong	(key, “sum,count”)
Distinct students per item	Union of ID sets, not counts	Two maps may see the same student	(item, student id)
Median, percentiles	Usually none	Needs all values (or a sketch structure)	(key, value)
Join, secondary sort	None	Nothing to pre-aggregate	—

Three ways to cut map output

Combiner

One line in the driver. Hadoop runs it on spills and merges. Saves shuffle bytes and disk, but every pair is still created and serialized first.

setCombinerClass

In-mapper combining

Keep a HashMap of totals in the mapper and write them in cleanup() (Lesson 8.3). Fewer objects and no repeated sorting; you must watch memory if there are many keys.

fastest

Filter early

Don't emit what the reducer will throw away: skip bad rows and unused columns in the mapper.

always

When none helps

If almost every key is unique (order IDs, timestamps), there is nothing to combine; the combiner only costs CPU.

check the counters

Read the ratio Combine output records ÷ Combine input records. Word count: 465 ÷ 20,665,132, a huge win. If the ratio is close to 1, the combiner is wasted work: remove it.

4 Step by step

A combiner that helps, and one that lies

Module 9 runs on the Module 8 lab cluster (Lesson 8.1) with a new, bigger dataset made by make_student_orders.py: 5 lakh orders by 2,000 students over September 2026, with the time of each order and 8 items including coffee and thali. Twenty “regulars” (S0001 to S0020) order far more than everyone else, which will matter for partitioning and skew.

Turn the combiner on
```
job.setMapperClass(TokenizerMapper.class);
if (conf.getBoolean("wc.combiner", true)) job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
```
The reducer itself works as the combiner because its input (Text, IntWritable) and output (Text, IntWritable) are both the map output types, and summing is safe to do in pieces.
Run word count without and with it
```
hadoop jar mr9.jar WordCountCombiner -D wc.combiner=false /user/hadoop/data/feedback_x20.txt /out/wc_nocomb
hadoop jar mr9.jar WordCountCombiner -D wc.combiner=true  /user/hadoop/data/feedback_x20.txt /out/wc_comb
```
== wc_nocomb 52918 ms Map output records=20664760 Combine input records=0 Reduce shuffle bytes=225867818 Reduce input records=20664760 Spilled Records=61994280 == wc_comb 29249 ms Map output records=20664760 Combine input records=20665132 Combine output records=465 Reduce shuffle bytes=1083 Reduce input records=93 Spilled Records=558
The maps did exactly the same work (20,664,760 pairs each time). What changed is everything after: 93 records instead of 2 crore reached the reducer, and the job was 45% faster.
Check the answer didn't change
```
diff <(hdfs dfs -cat /out/wc_nocomb/part-r-00000) <(hdfs dfs -cat /out/wc_comb/part-r-00000) && echo "identical output"
```
identical output
Always compare a run with and without the combiner once. A correct combiner never changes the output.
Notice the combiner ran twice
```
Combine input records  20,665,132 = 20,664,760 map pairs + 372
spill combines   : 20,664,760 pairs → 372 records (12 spills × 31 words)
merge combine    :        372       →  93 records (3 maps × 31 words)
Combine output   : 372 + 93 = 465
```
Each 102 MB map filled its sort buffer several times. The combiner ran on every spill, and because each map had at least 3 spills, it ran again while merging them. Your combiner must give the same result however many times it runs.
Now the dangerous one: average spend per student
```
hadoop jar mr9.jar AvgSpend -D avg.combiner=none  /user/hadoop/data/student_orders.csv /out/avg_none
hadoop jar mr9.jar AvgSpend -D avg.combiner=wrong /user/hadoop/data/student_orders.csv /out/avg_wrong
```
S0001 none: 25.90 (3899 orders) | wrong: 26.05 (3 orders) S0007 none: 25.64 (3934 orders) | wrong: 25.50 (3 orders) S1413 none: 25.13 (184 orders) | wrong: 25.30 (3 orders) -- students whose answer differs from the no-combiner run wrong: 2000 of 2000
The wrong combiner averaged each map's values and passed the average on as if it were a single order. The reducer then averaged three averages, one per map. No error, no warning: every one of the 2,000 answers is slightly wrong.
Fix it with a (sum, count) pair
```
// map emits "amount,1"; the combiner adds sums and counts, never divides
for (Text v : vals) { String[] p = v.toString().split(",");
  sum += Double.parseDouble(p[0]); n += Long.parseLong(p[1]); }
ctx.write(s, new Text(sum + "," + n));
```
== avg_right 23559 ms Combine input records=500000 Combine output records=6000 Reduce shuffle bytes=107140 S1413 none: 25.13 (184 orders) | right: 25.13 (184 orders) right: 0 of 2000
6,000 records (2,000 students × 3 maps) instead of 5 lakh, 61 times fewer shuffle bytes, and every answer correct. The division happens once, in the reducer.

Hadoop never checks whether your combiner is mathematically safe. A wrong combiner produces plausible numbers, not errors. Test with and without it on a sample, and compare the outputs.

5 See it happen

Inside one map task with a combiner

Six steps: spills get combined, merged spills get combined again, then the same idea applied to an average, wrongly and rightly.

Combine, merge, combine again 6 steps

What you see

Behind the screen

6 Real-world use

Combiners at work

1. Counting events in logs

Page views, errors or clicks per URL per hour: billions of (key, 1) pairs collapse to a few lakh partial totals per map before the shuffle.

(url, 1) × billions→combiner: sum→(url, partial total)

2. Metrics with partial aggregates

Average transaction value per merchant, average latency per API: the map emits (sum, count, min, max) so a combiner can merge them safely; the reducer finishes the averages.

Transactions→(sum, count) combiner→Averages per merchant

3. Inside SQL engines

Hive's hive.map.aggr and Spark's reduceByKey do the same thing automatically: a partial aggregation on the map side, then the final one after the shuffle. A GROUP BY with SUM or COUNT gets it for free.

GROUP BY item→partial aggregate→final aggregate

4. Approximate distinct counts

“Unique users per day” can't be added from partial counts. Teams combine HyperLogLog sketches instead: small structures that can be merged any number of times and give a close estimate.

(day, user)→merge sketches→≈ unique users

Simplified; organisations differ in tools and policies.

7 Run the code

Simulate a combiner running on spills

One map task's output is cut into spills, as the sort buffer would. The combiner runs on each spill, then once more on the merged spills. Compare the final answer with no combiner, a sum combiner and an averaging combiner.

pairs = [("chai", 15), ("chai", 15), ("dosa", 45), ("chai", 15), ("thali", 80),
         ("dosa", 45), ("chai", 15), ("thali", 80), ("chai", 20), ("dosa", 40)]
spills = [pairs[0:4], pairs[4:7], pairs[7:10]]        # the sort buffer filled 3 times

def combine(records, fn):                             # group by key, apply fn to each group
    groups = {}
    for k, v in records:
        groups.setdefault(k, []).append(v)
    return [(k, fn(vs)) for k, vs in sorted(groups.items())]

total = lambda vs: sum(vs)
avg = lambda vs: sum(vs) / len(vs)

for name, fn in [("sum", total), ("average", avg)]:
    after_spills = [r for s in spills for r in combine(s, fn)]   # combiner on every spill
    merged = combine(after_spills, fn)                           # and again on the merge
    print(f"{name:8} combiner ->", merged)
print("true sums      ->", combine(pairs, total))
print("true averages  ->", [(k, round(v, 2)) for k, v in combine(pairs, avg)])

sum combiner -> [('chai', 80), ('dosa', 130), ('thali', 160)] average combiner -> [('chai', 16.666666666666668), ('dosa', 43.333333333333336), ('thali', 80.0)] true sums -> [('chai', 80), ('dosa', 130), ('thali', 160)] true averages -> [('chai', 16.0), ('dosa', 43.33), ('thali', 80.0)]

spills = [pairs[0:4], pairs[4:7], pairs[7:10]] # the sort buffer filled 3 timesWith 3 or more spills Hadoop also runs the combiner while merging them, like our 102 MB maps did.
merged = combine(after_spills, fn) # and again on the mergeIts output becomes its own input: the function must give the same answer however often it runs.
avg = lambda vs: sum(vs) / len(vs)Average is not associative: averaging partial averages loses how many values each stood for.

Run it on your PC: with and without a combiner

Module 9 uses the Module 8 lab cluster (start it with docker compose -p lab8 up -d in your mr-lab folder, Lesson 8.1) and a new dataset: run make_student_orders.py to create student_orders.csv (5 lakh orders by 2,000 students, with time) and menu.csv. Copy both, the Java files and the script into /tmp/mr9 in the namenode container, compile everything into mr9.jar, and run the script. feedback_x20.txt is the 20× feedback file from Lesson 8.6. It takes about 3 minutes.

The Module 9 dataset

# make_student_orders.py: 5 lakh canteen orders by 2,000 students over 30 days (same file every time)
# Columns: order_id,student_id,time,item,amount      plus menu.csv: item,category,counter
import random
from datetime import datetime, timedelta
random.seed(21)
menu = {"chai": ("drink", 15, "A"), "coffee": ("drink", 20, "A"), "samosa": ("snack", 18, "B"),
        "vada pav": ("snack", 20, "B"), "poha": ("breakfast", 25, "C"), "idli": ("breakfast", 30, "C"),
        "dosa": ("meal", 45, "D"), "thali": ("meal", 80, "D")}
weights = [30, 8, 14, 14, 9, 9, 10, 6]                      # chai is by far the most ordered
items = list(menu)
students = [f"S{n:04d}" for n in range(1, 2001)]
heavy = students[:20]                                        # 20 students who practically live in the canteen
start = datetime(2026, 9, 1)
hours = list(range(8, 21))                                   # canteen open 08:00 to 20:59
hour_w = [9, 7, 4, 4, 5, 9, 10, 6, 4, 6, 8, 5, 3]          # rush at breakfast, lunch and evening tea
with open("student_orders.csv", "w") as f:
    for oid in range(1, 500_001):
        sid = random.choice(heavy) if random.random() < 0.15 else random.choice(students)
        t = start + timedelta(days=random.randrange(30), hours=random.choices(hours, hour_w)[0], minutes=random.randrange(60))
        item = random.choices(items, weights)[0]
        f.write(f"{oid},{sid},{t:%Y-%m-%d %H:%M},{item},{menu[item][1]}\n")
with open("menu.csv", "w") as f:
    for item, (cat, price, counter) in menu.items():
        f.write(f"{item},{cat},{price},{counter}\n")
print("student_orders.csv: 500,000 orders; menu.csv:", len(menu), "items")

# Run on your PC, in the mr-lab folder:
#   python make_student_orders.py
#   docker exec lab8-namenode-1 mkdir -p /tmp/mr9
#   docker cp student_orders.csv lab8-namenode-1:/tmp/mr9/
#   docker cp menu.csv lab8-namenode-1:/tmp/mr9/

student_orders.csv: 500,000 orders; menu.csv: 8 items $ head -3 student_orders.csv 1,S1413,2026-09-14 15:30,dosa,45 2,S0973,2026-09-26 20:11,samosa,18 3,S0007,2026-09-01 13:37,samosa,18

Real output. The random seed is fixed, so your file is identical to ours.

Word count with an optional combiner

// WordCountCombiner.java: word count with an optional combiner.  -D wc.combiner=true|false
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCountCombiner {
  public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private static final IntWritable ONE = new IntWritable(1);
    private final Text word = new Text();
    @Override
    public void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      StringTokenizer it = new StringTokenizer(line.toString().toLowerCase(), " \t.,!?;:\"'()");
      while (it.hasMoreTokens()) { word.set(it.nextToken()); ctx.write(word, ONE); }
    }
  }

  // Used as the reducer AND (optionally) as the combiner: summing is safe to do in pieces.
  public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private final IntWritable result = new IntWritable();
    @Override
    public void reduce(Text word, Iterable<IntWritable> counts, Context ctx) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable c : counts) sum += c.get();
      result.set(sum);
      ctx.write(word, result);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    Job job = Job.getInstance(conf, "word count combiner=" + conf.getBoolean("wc.combiner", true));
    job.setJarByClass(WordCountCombiner.class);
    job.setMapperClass(TokenizerMapper.class);
    if (conf.getBoolean("wc.combiner", true)) job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

// Copy all Module 9 .java files to /tmp/mr9 in the namenode container, then inside it:
//   cd /tmp/mr9
//   mkdir -p classes && javac -cp $(hadoop classpath) -d classes *.java && jar cf mr9.jar -C classes .
//   hadoop jar mr9.jar WordCountCombiner -D wc.combiner=false /user/hadoop/data/feedback_x20.txt /out/wc_nocomb

### Word count on feedback_x20.txt (102 MB, 3 maps) == wc_nocomb 52918 ms Launched map tasks=4 Map output records=20664760 Combine input records=0 Combine output records=0 Reduce shuffle bytes=225867818 Reduce input records=20664760 Spilled Records=61994280 == wc_comb 29249 ms Launched map tasks=3 Map output records=20664760 Combine input records=20665132 Combine output records=465 Reduce shuffle bytes=1083 Reduce input records=93 Spilled Records=558 -- same answer? identical output

Real output from combiner_lab.sh (below). The reducer doubles as the combiner because its input and output types are both (Text, IntWritable).

Average spend: no, wrong and right combiner

// AvgSpend.java: average amount per order for each student.
// -D avg.combiner=wrong   uses the averaging reducer as a combiner (gives WRONG answers)
// -D avg.combiner=right   map emits (sum,count) pairs and the combiner adds them (correct)
// -D avg.combiner=none    no combiner (correct, more data shuffled)
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class AvgSpend {
  // value travels as text "sum,count" so the same reducer can read map output and combiner output
  public static class M extends Mapper<LongWritable, Text, Text, Text> {
    private final Text student = new Text(), val = new Text();
    @Override
    protected void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      String[] f = line.toString().split(",");          // order_id,student_id,time,item,amount
      student.set(f[1]);
      val.set(f[4] + ",1");
      ctx.write(student, val);
    }
  }

  // WRONG as a combiner: it averages its input and claims the result is 1 order, so averages get averaged
  public static class AvgReducer extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text s, Iterable<Text> vals, Context ctx) throws IOException, InterruptedException {
      double sum = 0; long n = 0;
      for (Text v : vals) { String[] p = v.toString().split(","); sum += Double.parseDouble(p[0]); n += 1; }   // ignores the counts
      ctx.write(s, new Text((sum / n) + ",1"));                  // a partial AVERAGE, pretending to be 1 order
    }
  }

  // RIGHT as a combiner: adds sums and counts, never divides
  public static class SumCountCombiner extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text s, Iterable<Text> vals, Context ctx) throws IOException, InterruptedException {
      double sum = 0; long n = 0;
      for (Text v : vals) { String[] p = v.toString().split(","); sum += Double.parseDouble(p[0]); n += Long.parseLong(p[1]); }
      ctx.write(s, new Text(sum + "," + n));
    }
  }

  // Final reducer for the correct versions: divide once at the end
  public static class FinalAvg extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text s, Iterable<Text> vals, Context ctx) throws IOException, InterruptedException {
      double sum = 0; long n = 0;
      for (Text v : vals) { String[] p = v.toString().split(","); sum += Double.parseDouble(p[0]); n += Long.parseLong(p[1]); }
      ctx.write(s, new Text(String.format("%.2f (%d orders)", sum / n, n)));
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    String mode = conf.get("avg.combiner", "none");
    Job job = Job.getInstance(conf, "avg spend combiner=" + mode);
    job.setJarByClass(AvgSpend.class);
    job.setMapperClass(M.class);
    if (mode.equals("wrong")) job.setCombinerClass(AvgReducer.class);
    if (mode.equals("right")) job.setCombinerClass(SumCountCombiner.class);
    job.setReducerClass(FinalAvg.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

### Average spend per order, per student (student_orders.csv, 5 lakh orders) == avg_none 23359 ms Launched map tasks=3 Map output records=500000 Combine input records=0 Combine output records=0 Reduce shuffle bytes=6500018 Reduce input records=500000 Spilled Records=1000000 == avg_wrong 22617 ms Launched map tasks=3 Map output records=500000 Combine input records=500000 Combine output records=6000 Reduce shuffle bytes=161503 Reduce input records=6000 Spilled Records=12000 == avg_right 23559 ms Launched map tasks=3 Map output records=500000 Combine input records=500000 Combine output records=6000 Reduce shuffle bytes=107140 Reduce input records=6000 Spilled Records=12000

Real output from combiner_lab.sh. The value travels as the text "sum,count" so the reducer can read both map output and combiner output.

Run all five jobs and compare

# combiner_lab.sh: what a combiner saves, and how a wrong combiner silently breaks results
cd /tmp/mr9
hdfs dfs -D dfs.blocksize=8388608 -put -f student_orders.csv /user/hadoop/data/student_orders.csv
run() {  # name, then job args
  name=$1; shift
  hdfs dfs -rm -r -f -skipTrash /out/$name > /dev/null 2>&1
  start=$(date +%s%N)
  hadoop jar mr9.jar "$@" /out/$name > /tmp/$name.log 2>&1
  echo "== $name   $(( ($(date +%s%N) - start) / 1000000 )) ms"
  grep -E "Launched map tasks|Map output records|Combine input records|Combine output records|Reduce input records|Reduce shuffle bytes|Spilled Records" /tmp/$name.log | sed 's/^\s*/  /'
}
echo "### Word count on feedback_x20.txt (102 MB, 3 maps)"
run wc_nocomb WordCountCombiner -D wc.combiner=false /user/hadoop/data/feedback_x20.txt
run wc_comb   WordCountCombiner -D wc.combiner=true  /user/hadoop/data/feedback_x20.txt
echo "-- same answer?"
diff <(hdfs dfs -cat /out/wc_nocomb/part-r-00000) <(hdfs dfs -cat /out/wc_comb/part-r-00000) && echo "identical output"
echo
echo "### Average spend per order, per student (student_orders.csv, 5 lakh orders)"
run avg_none  AvgSpend -D avg.combiner=none  /user/hadoop/data/student_orders.csv
run avg_wrong AvgSpend -D avg.combiner=wrong /user/hadoop/data/student_orders.csv
run avg_right AvgSpend -D avg.combiner=right /user/hadoop/data/student_orders.csv
echo "-- three students, three runs"
for s in S0001 S0007 S1413; do
  echo "$s  none: $(hdfs dfs -cat /out/avg_none/part-r-00000 | grep "^$s" | cut -f2)  |  wrong: $(hdfs dfs -cat /out/avg_wrong/part-r-00000 | grep "^$s" | cut -f2)  |  right: $(hdfs dfs -cat /out/avg_right/part-r-00000 | grep "^$s" | cut -f2)"
done
echo "-- students whose answer differs from the no-combiner run"
echo "wrong: $(diff <(hdfs dfs -cat /out/avg_none/part-r-00000) <(hdfs dfs -cat /out/avg_wrong/part-r-00000) | grep -c '^>') of 2000"
echo "right: $(diff <(hdfs dfs -cat /out/avg_none/part-r-00000) <(hdfs dfs -cat /out/avg_right/part-r-00000) | grep -c '^>') of 2000"

# Inside the namenode container:  bash /tmp/mr9/combiner_lab.sh

-- three students, three runs S0001 none: 25.90 (3899 orders) | wrong: 26.05 (3 orders) | right: 25.90 (3899 orders) S0007 none: 25.64 (3934 orders) | wrong: 25.50 (3 orders) | right: 25.64 (3934 orders) S1413 none: 25.13 (184 orders) | wrong: 25.30 (3 orders) | right: 25.13 (184 orders) -- students whose answer differs from the no-combiner run wrong: 2000 of 2000 right: 0 of 2000

Real output: the last part of the script, comparing the three average runs (the job counters are shown with the two Java files above). “Launched map tasks=4” in the first word count run is a speculative extra attempt (Lesson 8.1), not a 4th split. Your timings will differ; the record counts and the 2000-of-2000 result will not.

8 Line by line

The combiner runs twice inside one map task

Two spills are combined, then merged and combined again. Watch out after each pass.

Worked examples: real problems, solved and traced in memory

Three combiner problems: estimating the saving, deciding whether a function is safe, and designing a partial aggregate.

Example 1BeginnerSizing · shuffle

How much does the combiner save?

Problem statement

Our word count job emitted 20,664,760 pairs and shuffled 225,867,818 bytes without a combiner. With it, the reducer received 93 records in 1,083 bytes. Compute bytes per record in both runs and the reduction factor.

Input

20,664,760 pairs / 225,867,818 B · 93 records / 1,083 B

Expected output

10.9 B/record before, 11.6 B/record after
208,558x less data shuffled

Approach

Bytes per record = bytes / records.
Reduction = bytes before / bytes after.
Compare.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

Example 2IntermediateCorrectness · rule

Which functions are safe to combine?

Problem statement

Test sum, max, average and “count” (number of values) by combining a list in two different groupings and checking whether the results agree with applying the function once to the whole list.

Input

values 15, 20, 45, 80, 15

Expected output

sum   once=175    combined=175    safe=True
max   once=80     combined=80     safe=True
avg   once=35.0   combined=32.08333333333333 safe=False
count once=5      combined=2      safe=False

Approach

Apply f to the whole list.
Split into two parts, apply f to each, then f to the results.
Safe if both ways agree.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

Example 3AdvancedDesign · partials

Design a combiner for min, max and average together

Problem statement

The canteen wants per student: number of orders, total, average, smallest and largest order. Design one partial aggregate that a combiner can merge safely, and finish it in the reducer.

Input

two partial results for S1413

Expected output

184 orders, total 4624, avg 25.13, min 15, max 80

Approach

Partial = (count, sum, min, max): all merge with +, +, min, max.
Combiner merges partials field by field.
Reducer computes average = sum / count at the end.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

9 Quiz and revision

Check yourself

Quick revision

Lesson 9.1 in one screen

Combiner = mini-reducer on one map task's output, before the shuffle; job.setCombinerClass(...).
Runs on each spill and again on merges (≥ 3 spills, mapreduce.map.combine.minspills); may run 0, 1 or many times.
Must be commutative and associative; input type = output type = map output type.
Lab word count: 225,867,818 → 1,083 shuffle bytes, 52.9 s → 29.2 s, identical output; Combine input 20,665,132 > Map output (ran twice).
Safe: sum, count (as sum of 1s), max, min. Unsafe as-is: average, median, distinct count.
Average: emit (sum, count), divide only in the reducer. Wrong combiner → 2,000 of 2,000 averages wrong, silently.
Alternative: in-mapper combining (HashMap + cleanup()). Useless when keys are nearly unique.

Interview and viva questions

What is a combiner in MapReduce?

A combiner is an optional local reducer that runs on the output of each map task before the shuffle. It aggregates values for the same key on the map side, so far less data is written to disk and sent over the network to the reducers. It is set with job.setCombinerClass() and often reuses the reducer class when the reduce function is commutative and associative.

Is the combiner guaranteed to run?

No. Hadoop may run the combiner zero, one or several times for a map task: on each spill of the sort buffer, again while merging spill files (when there are at least 3 spills), and sometimes during reduce-side merges. The job must produce correct results whether or not it runs.

Can every reducer be used as a combiner?

No. Only when the reduce function is commutative and associative and its input and output types are the same as the map output types. Sums, counts expressed as sums, max and min work. Averages, medians and distinct counts do not; for an average, emit (sum, count) pairs and divide only in the reducer.

What is the difference between a combiner and a reducer?

A reducer receives all values of a key from every map task and produces the final output, which is written to HDFS. A combiner sees only part of the values (one map task's, often one spill's), its output goes into the shuffle rather than HDFS, and it may run any number of times, so it can only do work that is safe to repeat on partial data.

What is in-mapper combining?

Aggregating inside the mapper itself, for example in a HashMap of key → running total, and writing the totals once in cleanup(). It avoids creating and sorting every intermediate pair, so it is usually faster than a combiner, but it uses mapper memory and you must flush the map if it grows too large.

Practice problems

Combiner exercises using the real counters from our lab.

P1Easy

Time saved

Word count took 52,918 ms without a combiner and 29,249 ms with one. What percentage of time was saved?

Input

52,918 ms vs 29,249 ms

Expected output

45% faster

Show solution and explanation

print(f"{(52918 - 29249) / 52918 * 100:.0f}% faster")

45% faster

How it works: About 45% less time, from shuffling 93 records instead of 2 crore.

P2Easy

Records per map

The reducer received 93 records from 3 map tasks. How many distinct words are there?

Input

93 records, 3 maps

Expected output

31 distinct words

Show solution and explanation

print(93 // 3, "distinct words")

31 distinct words

How it works: 31: each map sends one combined record per word, and every map saw all 31 words.

P3Easy

Partial averages

With the right combiner the reducer got 6,000 records for 2,000 students. Why 6,000?

Input

6,000 records, 2,000 students

Expected output

3.0 partial (sum, count) records per student = number of map tasks

Show solution and explanation

print(6000 / 2000, "partial (sum, count) records per student = number of map tasks")

3.0 partial (sum, count) records per student = number of map tasks

How it works: 3 per student: one partial from each of the 3 map tasks.

P4Medium

The second combine pass

Combine input was 20,665,132 and map output 20,664,760. Combine output was 465 and the reducer got 93. Work out how many records the spill passes produced.

Input

four counters

Expected output

records combined a second time: 372
spill passes produced: 372 -> merge pass produced: 93
spills in total: 12

Show solution and explanation

extra_input = 20665132 - 20664760
print("records combined a second time:", extra_input)
print("spill passes produced:", 465 - 93, "-> merge pass produced:", 93)
print("spills in total:", extra_input // 31)

records combined a second time: 372 spill passes produced: 372 -> merge pass produced: 93 spills in total: 12

How it works: 372 records from 12 spills (31 words each) were combined again into 93 during the merges.

P5Medium

Recreate the wrong average

S1413 had 184 orders. Suppose maps saw 70, 52 and 62 orders with averages 25.4, 24.1 and 25.7. Compute the wrong (average of averages) and the right (weighted) answer.

Input

3 partial averages and counts

Expected output

wrong 25.07  right 25.13

Show solution and explanation

parts = [(25.4, 70), (24.1, 52), (25.7, 62)]
wrong = sum(a for a, n in parts) / len(parts)
right = sum(a * n for a, n in parts) / sum(n for a, n in parts)
print(f"wrong {wrong:.2f}  right {right:.2f}")

wrong 25.07 right 25.13

How it works: The unweighted average gives each map equal weight, whatever it saw. Weighting by count fixes it. (Partials illustrative.)

P6Medium

Distinct students, safely

Combine partial sets of student IDs instead of partial counts, and count only at the end.

Input

2 maps

Expected output

3 distinct students

Show solution and explanation

map1, map2 = {"S0001", "S0002"}, {"S0001", "S0003"}
combined = map1 | map2               # union is commutative and associative
print(len(combined), "distinct students")

3 distinct students

How it works: 3. Set union is safe to combine; adding counts is not. With millions of IDs, teams use HyperLogLog sketches instead of full sets.

P7Business problem

Network bill

A nightly job shuffles 4 TB between racks without a combiner; with one it shuffles 40 GB. If cross-rack transfer is the bottleneck at 10 Gbit/s, how long does the shuffle take each way?

Input

4 TB vs 40 GB at 10 Gbit/s

Expected output

no combiner  53.3 minutes
combiner     0.5 minutes

Show solution and explanation

rate = 10e9 / 8                         # bytes per second
for name, b in [("no combiner", 4e12), ("combiner", 40e9)]:
    print(f"{name:12} {b / rate / 60:.1f} minutes")

no combiner 53.3 minutes combiner 0.5 minutes

How it works: About 53 minutes vs half a minute of pure transfer (illustrative). The combiner also saves the disk writes on both sides.

P8Advanced

In-mapper combining

Instead of a combiner, aggregate inside the mapper with a dictionary and emit totals at the end (cleanup). Flush when the dictionary exceeds 3 keys, to bound memory. How many records are emitted?

Input

10 words, flush above 3 keys

Expected output

7 records emitted for 10 words: [('chai', 2), ('was', 2), ('hot', 1), ('cold', 1), ('chai', 2), ('was', 1), ('late', 1)]

Show solution and explanation

words = "chai was hot chai was cold chai was late chai".split()
buf, emitted = {}, []
for w in words:
    buf[w] = buf.get(w, 0) + 1
    if len(buf) > 3:                     # memory guard: flush partial totals
        emitted += buf.items(); buf = {}
emitted += buf.items()                   # cleanup()
print(len(emitted), "records emitted for", len(words), "words:", emitted)

7 records emitted for 10 words: [('chai', 2), ('was', 2), ('hot', 1), ('cold', 1), ('chai', 2), ('was', 1), ('late', 1)]

How it works: 7 records for 10 words: one flush happened, so chai and was are emitted twice. Without the memory limit it would be 5 (one per word). The reducer still sums correctly because partial totals for the same word are simply added.

Real-world scenario questions (asked in company interviews)

Combiners are a favourite interview topic because one wrong answer reveals whether a candidate understands what “local aggregation” really means.

Concepts · basicsFresher

“What is a combiner and why is it used?”

Asked right after word count.

The interviewer is checking: where it runs and what it saves.

Show a strong answer

A local mini-reducer that runs on each map task's output, set with setCombinerClass.
It reduces the data written to local disk and sent in the shuffle, e.g. (chai, 1) × lakhs → (chai, total).
In our lab word count: 225 MB of shuffle became 1 KB and the job ran 45% faster.

Correctness · designFresher

“Can I use my average reducer as a combiner?”

The classic trick question.

The interviewer is checking: associativity and partial aggregates.

Show a strong answer

No: an average of partial averages is wrong unless every part has the same count, and Hadoop may run the combiner any number of times.
Emit (sum, count) from the map, let the combiner add both, and divide once in the reducer.
The combiner's input and output types must both match the map output types, which (sum, count) does.

Debugging · productionExperienced

Results changed slightly after a “performance fix”

A colleague added a combiner to a metrics job; dashboards now differ from last month by 0.5–2%. No job failed.

The interviewer is checking: silent correctness bugs.

Show a strong answer

Suspect a non-associative combiner (average, ratio, percentile, distinct count).
Rerun the same input with and without the combiner and diff the outputs; check Combine input/output counters.
Fix with partial aggregates (sum, count; min, max; sets or sketches) and add a regression test comparing both modes on a sample.

Performance · tuningExperienced

“We added a combiner but the job didn't get faster”

Counters show Combine input 50 crore, Combine output 49 crore.

The interviewer is checking: reading counters.

Show a strong answer

The ratio is about 0.98: keys are almost unique per map task, so there is nothing to combine; the combiner only adds CPU.
Remove it, or change the key design (coarser keys) if aggregation is the goal.
Look elsewhere: compress map output, filter earlier, or use in-mapper combining for hot keys.

Combiner

Two crore ones on the network

Combiner

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, Module 8 lab cluster)

The combiner at three levels

Beginner Tally sheets at counting tables

Intermediate Where it runs and what the counters say

Advanced What is safe to combine

Which jobs can use a combiner?

Three ways to cut map output

Combiner

In-mapper combining

Filter early

When none helps

A combiner that helps, and one that lies

Turn the combiner on

Run word count without and with it

Check the answer didn't change

Notice the combiner ran twice

Now the dangerous one: average spend per student

Fix it with a (sum, count) pair

Inside one map task with a combiner

Combiners at work

1. Counting events in logs

2. Metrics with partial aggregates

3. Inside SQL engines

4. Approximate distinct counts

Simulate a combiner running on spills

Run it on your PC: with and without a combiner

The Module 9 dataset

Word count with an optional combiner

Average spend: no, wrong and right combiner

Run all five jobs and compare

The combiner runs twice inside one map task

Worked examples: real problems, solved and traced in memory

How much does the combiner save?

Which functions are safe to combine?

Design a combiner for min, max and average together

Check yourself

Quick revision

Lesson 9.1 in one screen

Interview and viva questions

Practice problems

Time saved

Records per map

Partial averages

The second combine pass

Recreate the wrong average

Distinct students, safely

Network bill

In-mapper combining

Real-world scenario questions (asked in company interviews)

“What is a combiner and why is it used?”

“Can I use my average reducer as a combiner?”

Results changed slightly after a “performance fix”

“We added a combiner but the job didn't get faster”