Data Intensive Text Processing with MapReduce #3 figure3.9.x Reducer

gistfile1.java

package info.moaikids.mapred.reduce;

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

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Reducer;

public class Figure39xReducer extends
		Reducer<Text, MapWritable, Text, DoubleWritable> {

	static final int MIN = 1;

	@Override
	protected void reduce(Text key, Iterable<MapWritable> values,
			Context context) throws IOException, InterruptedException {
		Map<String, Integer> map = new HashMap<String, Integer>();
		int total = 0;
		for (MapWritable value : values) {
			for (Entry<Writable, Writable> entry : value.entrySet()) {
				String text = ((Text) entry.getKey()).toString();
				int count = ((IntWritable) entry.getValue()).get();
				
				total += count;
				if (map.containsKey(text)) {
					map.put(text, map.get(text) + count);
				} else {
					map.put(text, count);
				}
			}
		}

		for (Entry<String, Integer> entry : map.entrySet()) {
			if (entry.getValue() > MIN) {
				context.write(new Text(key.toString() + " " + entry.getKey()),
						new DoubleWritable(((double)entry.getValue()) / total));
			}

		}

	}
}