Hadoop Reducer Iterable: How to Iterate Twice

By roman10 On August 13, 2012 · 1 Comment

The reducer of Hadoop MapReduce program receives an Iterable of inputs with the same key. Most of the times, we iterate through the values once. But there’re always exceptions. What if we want to iterate the values twice?

The code shown below are tested on Hadoop 1.0.3.

0. An example

We’ll use a simplified version of the example in How to Write a Simple MapReduce Program. Suppose we are given several files, each file contains 5 key value pairs. The key indicates a year in the range of [2000,2001], and the value is some random number generated. The task is to find the maximum number for each year. Note that for each file, it may contain values for multiple years.

Below is a script that helps us to generate the input files.

#!/bin/bash
for i in {1..5}
do
echo $(($RANDOM%2+2000)) $RANDOM
done

To generate the input, use the commands below.

$ mkdir input
$ ./rand.sh > input/1.txt
$ ./rand.sh > input/2.txt

1. The wrong code
One might think that iterating the Iterable is easy. We simply iterate the Iterable twice. Like the code below,

private static class MaxNumReducerWrong1 extends Reducer<Text, LongWritable, Text, LongWritable> {
        @Override
        public void reduce(Text pKey, Iterable<LongWritable> pValues, Context pContext) 
                throws IOException, InterruptedException {
            System.out.println("----------" + pKey + "----------");
              for (LongWritable aNum : pValues) {
                    System.out.println("first iteration: " + aNum);
                }
                for (LongWritable aNum : pValues) {
                    System.out.println("second iteration: " + aNum);
                }
          }
}

However, this won’t work. The output we’ll get are similar to below.

Basically, the second loop is never executed, in other words, the Iterable becomes empty after the first iteration.

One can try using two Iterators. The result is the same.

2. Another Wrong Code

The way to iterate through the Iterable twice is to cache the values, like the code below.

private static class MaxNumReducerWrong2 extends Reducer<Text, LongWritable, Text, LongWritable> {
        @Override
        public void reduce(Text pKey, Iterable<LongWritable> pValues, Context pContext) 
                throws IOException, InterruptedException {
            System.out.println("----------" + pKey + "----------");
            ArrayList<LongWritable> cache = new ArrayList<LongWritable>();
              for (LongWritable aNum : pValues) {
                System.out.println("first iteration: " + aNum);
                cache.add(aNum);
            }
              int size = cache.size();
            for (int i = 0; i < size; ++i) {
                System.out.println("second iteration " + i + ": " + cache.get(i));
            }
          }
    }

But this doesn’t work also. We’ll get the output as below.
This indicates the Iterable is giving us the same object reference every time, but with different values. Since we’re caching the reference, all cached reference will be pointing to the same object. At the end of the iteration, the object contains the value of last element. Therefore, our cached references point to the same object, which has the value of the last element.

3. The Code Works

Basically, we’ll need to cache the objects. Below is the code that works.

private static class MaxNumReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
        @Override
        public void reduce(Text pKey, Iterable<LongWritable> pValues, Context pContext) 
                throws IOException, InterruptedException {
            System.out.println("----------" + pKey + "----------");
            ArrayList<LongWritable> cache = new ArrayList<LongWritable>();
            for (LongWritable aNum : pValues) {
                System.out.println("first iteration: " + aNum);
                LongWritable writable = new LongWritable();
                writable.set(aNum.get());
                cache.add(writable);
            }
            int size = cache.size();
            for (int i = 0; i < size; ++i) {
                System.out.println("second iteration: " + cache.get(i));
            }
        }
    }

And there is another version using Iterator,

private static class MaxNumReducer2 extends Reducer<Text, LongWritable, Text, LongWritable> {
        @Override
        public void reduce(Text pKey, Iterable<LongWritable> pValues, Context pContext) 
                throws IOException, InterruptedException {
            System.out.println("----------" + pKey + "----------");
            ArrayList<LongWritable> cache = new ArrayList<LongWritable>();
            Iterator<LongWritable> iterator = pValues.iterator();
            while (iterator.hasNext()) {
                LongWritable writable = iterator.next();
                System.out.println("MaxNumReducer2 first iteration: " + writable);
                cache.add(new LongWritable(writable.get()));
            }
            int size = cache.size();
            for (int i = 0; i < size; ++i) {
                System.out.println("MaxNumReducer2 second iteration: " + cache.get(i));
            }
        }
    }