Data Intensive Text Processing with MapReduce #3 SecondarySort (Reduce Side) Reducer

gistfile1.java

package info.moaikids.mapred.reduce;

import java.io.IOException;
import java.util.Map.Entry;
import java.util.TreeMap;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class ReduceSecondarySortReducer extends
		Reducer<Text, Text, Text, IntWritable> {

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		TreeMap<String, Integer> map = new TreeMap<String, Integer>();
		for (Text value : values) {
			String tsv = value.toString().trim();
			if (tsv.isEmpty()) {
				continue;
			}
			String[] params = tsv.split("\t");
			String year = params[0].trim();
			if (map.containsKey(year)) {
				map.put(year, map.get(year) + 1);
			} else {
				map.put(year, 1);
			}
		}


		for (Entry<String, Integer> entry : map.entrySet()) {
			context.write(new Text(key.toString() + " " + entry.getKey()),
					new IntWritable(entry.getValue()));
		}
	}

}