The Reducer

Three desks, one of them empty

The Reducer

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, 3 workers, canteen_orders.csv)

The Reducer at three levels

Choosing the number of reducers

What a reducer typically does

Reducers on a real cluster

All runs use the Module 8 lab and reducer_lab.sh (in “Run it on your PC”). Input: canteen_orders.csv, 3 lakh orders, 6 items.

1. The reducer we are running

   public static class SumReducer extends Reducer<Text, IntWritable, Text, Text> {
     protected void reduce(Text item, Iterable<IntWritable> amounts, Context ctx) … {
       long orders = 0, revenue = 0;
       for (IntWritable a : amounts) { orders++; revenue += a.get(); }
       ctx.write(item, new Text(orders + "\t" + revenue));
     }
   }

   The four type parameters: input key Text, input value IntWritable (they must match the map output), output key Text, output value Text. a.get() copies the number out at once, so object reuse does no harm here.

2. Run with 3 reducers

   hadoop jar mr.jar ItemRevenue -D mapreduce.job.reduces=3 /user/hadoop/data/canteen_orders.csv /out/red_3

   Launched reduce tasks=3 Reduce input groups=6 Reduce output records=6 part-r-00000 (0 bytes): part-r-00001 (79 bytes): chai dosa samosa vada pav part-r-00002 (38 bytes): idli poha
   The default HashPartitioner sent chai, dosa, samosa and vada pav to partition 1, idli and poha to 2, and nothing to 0. Reducer 0 still started, found no keys and wrote an empty file. Each file is sorted on its own.

3. Check the partitions yourself

   def text_hash(s):                     # Hadoop Text.hashCode()
       h = 1
       for b in s.encode():
           h = (31 * h + b) & 0xffffffff
       return h
   for item in ["chai", "dosa", "idli", "poha", "samosa", "vada pav"]:
       print(item, (text_hash(item) & 0x7fffffff) % 3)

   chai 1 dosa 1 idli 2 poha 2 samosa 1 vada pav 1
   Exactly the files Hadoop wrote. The partition depends only on the key and the number of reducers, so the same key always lands on the same reducer.

4. Run with 10 reducers

   hadoop jar mr.jar ItemRevenue -D mapreduce.job.reduces=10 /user/hadoop/data/canteen_orders.csv /out/red_10

   Launched reduce tasks=11 part-r-00000 (0 bytes): part-r-00001 (0 bytes): part-r-00002 (22 bytes): vada pav part-r-00003 (0 bytes): part-r-00004 (19 bytes): dosa part-r-00005 (19 bytes): poha part-r-00006 (18 bytes): chai part-r-00007 (20 bytes): samosa part-r-00008 (0 bytes): part-r-00009 (19 bytes): idli
   Ten tasks, ten files, four empty: you can never use more reducers than you have keys. “Launched reduce tasks=11” is one extra speculative attempt, the same mechanism we saw for maps in Lesson 8.1.

5. Run with 0 reducers

   hadoop jar mr.jar ItemRevenue -D mapreduce.job.reduces=0 /user/hadoop/data/canteen_orders.csv /out/red_0
   hdfs dfs -cat /out/red_0/part-m-00000 | head -3

   Launched map tasks=1 2699986 /out/red_0/part-m-00000 samosa 18 vada pav 20 poha 25
   The reducer class is ignored. Map output goes straight to HDFS, in input order, one pair per order: the shuffle, sort and reduce simply don't happen.

6. The object-reuse trap

   for (Text id : ids) {
     first3.add(copy ? new Text(id) : id);   // without a copy: the SAME object 3 times
     if (first3.size() == 3) break;
   }

   -- copy (correct) -- no copy (-D sample.copy=false) chai [104221, 150350, 113456] chai [113456, 113456, 113456] dosa [78778, 213441, 200658] dosa [200658, 200658, 200658] idli [101007, 50991, 101001] idli [101001, 101001, 101001]
   The iterator reuses one Text; after three steps it holds the third ID, and all three list entries point to it. Notice too that the correct IDs are not in order: values arrive in no defined order within a key.

Three reducers and one reused object

Six steps: keys are partitioned, each reducer is called per key, files are committed, then the object-reuse bug in slow motion.

Reducers at work

Simplified; organisations differ in tools and policies.

Partition, group and reduce like Hadoop

This model deals keys to reducers with Hadoop's real Text hash, groups values per key, and calls a reduce function once per key in sorted order. Its partitions match the files our cluster wrote.

pairs = [("samosa", 18), ("vada pav", 20), ("poha", 25), ("samosa", 18), ("chai", 15),
         ("idli", 30), ("dosa", 45), ("chai", 15), ("poha", 25), ("idli", 30)]

def text_hash(s):                         # Hadoop's Text.hashCode()
    h = 1
    for b in s.encode():
        h = (31 * h + b) & 0xffffffff
    return h

def partition(key, n):                    # HashPartitioner
    return (text_hash(key) & 0x7fffffff) % n

def reduce_fn(key, values):
    return f"{key}\t{len(values)}\t{sum(values)}"

R = 3
parts = [{} for _ in range(R)]
for k, v in pairs:
    parts[partition(k, R)].setdefault(k, []).append(v)

for r in range(R):
    lines = [reduce_fn(k, parts[r][k]) for k in sorted(parts[r])]
    print(f"part-r-{r:05d}:", lines if lines else "(empty)")

• return (text_hash(key) & 0x7fffffff) % nThe HashPartitioner: clear the sign bit, then modulo the number of reducers. Same key, same reducer, always.
• lines = [reduce_fn(k, parts[r][k]) for k in sorted(parts[r])]Each reducer handles its keys in sorted order, one reduce() call per key.
• print(f"part-r-{r:05d}:", lines if lines else "(empty)")Reducer 0 gets no keys but still writes a file: our part-r-00000 was 0 bytes.

Run it on your PC: number of reducers and the reuse trap

Use the Module 8 lab from Lesson 8.1, with canteen_orders.csv in HDFS and mr.jar built from ItemRevenue.java and SampleOrders.java. The script runs ItemRevenue with 3, 10 and 0 reducers, lists every output file with its size and keys, then runs SampleOrders with and without copying values. It takes about 3 minutes.

// SampleOrders.java: for each item, list the first 3 order ids the reducer sees.
// Run with -D sample.copy=false to see the famous "object reuse" bug.
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
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 SampleOrders {
  public static class M extends Mapper<LongWritable, Text, Text, Text> {
    private final Text item = new Text(), id = new Text();
    @Override
    protected void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      String[] f = line.toString().split(",");
      if (f.length != 3 || f[0].equals("order_id")) return;
      item.set(f[1]); id.set(f[0]);
      ctx.write(item, id);
    }
  }

  public static class R extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text item, Iterable<Text> ids, Context ctx) throws IOException, InterruptedException {
      boolean copy = ctx.getConfiguration().getBoolean("sample.copy", true);
      List<Text> first3 = new ArrayList<>();
      for (Text id : ids) {
        first3.add(copy ? new Text(id) : id);       // without a copy we keep the SAME object 3 times
        if (first3.size() == 3) break;
      }
      ctx.write(item, new Text(first3.toString()));
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    Job job = Job.getInstance(conf, "sample orders");
    job.setJarByClass(SampleOrders.class);
    job.setMapperClass(M.class);
    job.setReducerClass(R.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);
  }
}

// Inside the namenode container, in /tmp/mr:
//   javac -cp $(hadoop classpath) -d classes *.java && jar cf mr.jar -C classes .
//   hadoop jar mr.jar SampleOrders /user/hadoop/data/canteen_orders.csv /out/sample_ok
//   hadoop jar mr.jar SampleOrders -D sample.copy=false /user/hadoop/data/canteen_orders.csv /out/sample_bug

# reducer_lab.sh: number of reducers = number of output files; which key goes where; the object-reuse trap
cd /tmp/mr
D=/user/hadoop/data/canteen_orders.csv
for r in 3 10; do
  hdfs dfs -rm -r -f -skipTrash /out/red_$r > /dev/null 2>&1
  hadoop jar mr.jar ItemRevenue -D mapreduce.job.reduces=$r $D /out/red_$r 2>&1 | grep -E "Launched reduce tasks|Reduce input groups|Reduce output records"
  echo "== $r reducers: output files"
  for f in $(hdfs dfs -ls /out/red_$r | grep part-r | awk '{print $8}'); do
    echo "$(basename $f) ($(hdfs dfs -stat %b $f) bytes): $(hdfs dfs -cat $f | cut -f1 | tr '\n' ' ')"
  done
done
echo "== 0 reducers (map-only): mapper output written directly, unsorted"
hdfs dfs -rm -r -f -skipTrash /out/red_0 > /dev/null 2>&1
hadoop jar mr.jar ItemRevenue -D mapreduce.job.reduces=0 $D /out/red_0 2>&1 | grep -E "Launched (map|reduce)"
hdfs dfs -ls /out/red_0 | grep part | awk '{print $5, $8}'
hdfs dfs -cat /out/red_0/part-m-00000 2>/dev/null | head -3
echo "== first 3 order ids per item, copying each value (correct)"
hdfs dfs -rm -r -f -skipTrash /out/sample_ok /out/sample_bug > /dev/null 2>&1
hadoop jar mr.jar SampleOrders $D /out/sample_ok > /dev/null 2>&1
hdfs dfs -cat /out/sample_ok/part-r-00000
echo "== the same job WITHOUT copying (-D sample.copy=false)"
hadoop jar mr.jar SampleOrders -D sample.copy=false $D /out/sample_bug > /dev/null 2>&1
hdfs dfs -cat /out/sample_bug/part-r-00000

# Copy into the namenode container and run:
#   docker cp reducer_lab.sh lab8-namenode-1:/tmp/mr/
#   docker exec lab8-namenode-1 bash /tmp/mr/reducer_lab.sh

How a reducer walks through its keys

A reducer's input is one sorted stream of (key, value) pairs. Grouping means: keep reading while the key stays the same, then call reduce for that key.

Worked examples: real problems, solved and traced in memory

Three reducer problems: choosing the count, predicting output files, and spotting skew.

20 nodes × 6 containers

one wave : 114
two waves: 210

6 items · 4 reducers

part-r-00000 []
part-r-00001 ['idli', 'poha']
part-r-00002 ['chai', 'dosa', 'vada pav']
part-r-00003 ['samosa']

49 × 2 lakh records · 1 × 9 crore records · 60 min

hot reducer has 450x the normal load
even spread: 1,996,000 records each, about 1.3 min

Check yourself

Quick revision

Interview and viva questions

Practice problems

Reducer exercises using the real files and counters from our lab.

4 output lines

79 bytes

lines = ["chai\t50170\t752550", "dosa\t49948\t2247660", "samosa\t50021\t900378", "vada pav\t49986\t999720"]
print(sum(len(l) + 1 for l in lines), "bytes")

6 keys · 10 reducers

4 at least; more if two keys share a reducer

print(10 - 6, "at least; more if two keys share a reducer")

300,000 / 6

50000.0

print(300000 / 6)

iterator of amounts

('dosa', 4, 175, 45)

def reduce_fn(key, values):
    n = s = 0
    m = None
    for v in values:
        n += 1; s += v
        m = v if m is None or v > m else m
    return key, n, s, m
print(reduce_fn("dosa", iter([45, 45, 40, 45])))

sorted pairs

chai 30
idli 30
poha 50

from itertools import groupby
stream = [("chai", 15), ("chai", 15), ("idli", 30), ("poha", 25), ("poha", 25)]
for key, group in groupby(stream, key=lambda kv: kv[0]):
    print(key, sum(v for _, v in group))

“upma”, 3 reducers

part-r-00001

h = 1
for b in "upma".encode():
    h = (31 * h + b) & 0xffffffff
print("part-r-%05d" % ((h & 0x7fffffff) % 3))

400 reducers · 8 GB/day · 365 days

146,000 files a year, 20 MB each
better reducer count: 32

files = 400 * 365
print(f"{files:,} files a year, {8 * 1024 / 400:.0f} MB each")
print("better reducer count:", 8 * 1024 // 256)

record counts per key

largest group before: 900000 after: 225000
chai recombined: 900000

counts = {"chai": 900_000, "dosa": 50_000, "idli": 50_000, "poha": 50_000}
salted = {}
for k, c in counts.items():
    parts = 4 if k == "chai" else 1
    for i in range(parts):
        key = f"{k}#{i}" if parts > 1 else k
        salted[key] = c // parts
print("largest group before:", max(counts.values()), "after:", max(salted.values()))
total_chai = sum(v for k, v in salted.items() if k.startswith("chai#"))
print("chai recombined:", total_chai)

Real-world scenario questions (asked in company interviews)

Reducer questions test whether you have actually written MapReduce code: partitioning, reducer count and object reuse come up again and again.