求亿级记录中搜索次数Top N的搜索词(MapReduce实现)

程序事例：

日志信息：

二手车    1345
二手房    3416
洗衣机    2789

输入： N=2

输出：
二手房
洗衣机

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

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TopNMapper extends Mapper<Object, Text, NullWritable, Text> {
    private TreeMap<IntWritable, Text> tm = new TreeMap<IntWritable, Text>();
    private IntWritable mykey = new IntWritable();
    private Text myvalue = new Text();
    private int N = 10;
    @Override
    protected void map(Object key, Text value,
            Mapper<Object, Text, NullWritable, Text>.Context context)
            throws IOException, InterruptedException {
        String word = value.toString().split("\t")[0];
        int num = Integer.parseInt(value.toString().split("\t")[1]);
        mykey.set(num);
        myvalue.set(word);
        tm.put(mykey, myvalue);
        if (tm.size() > N) tm.remove(tm.firstKey());
    }
    
    @Override
    protected void cleanup(
            Mapper<Object, Text, NullWritable, Text>.Context context)
            throws IOException, InterruptedException {
        for (Map.Entry<IntWritable, Text> entry : tm.entrySet()) {
            Text value = new Text(entry.getKey() + " " + entry.getValue());
            context.write(NullWritable.get(), value);
        }     
    }
}

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

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

public class TopNReducer extends Reducer<NullWritable, Text, NullWritable, Text>{
    private TreeMap<IntWritable, Text> tm = new TreeMap<IntWritable, Text>();
    private IntWritable mykey = new IntWritable();
    private Text myvalue = new Text();
    private int N = 10;
    @Override
    protected void reduce(NullWritable key, Iterable<Text> values,
            Reducer<NullWritable, Text, NullWritable, Text>.Context context)
            throws IOException, InterruptedException {
        for (Text val : values) {
            String[] tmp = val.toString().split(" ");
            mykey.set(Integer.parseInt(tmp[0]));
            myvalue.set(tmp[1]);
            tm.put(mykey, myvalue);
            if (tm.size() > N) tm.remove(tm.firstKey());
        }
        for (Text res : tm.descendingMap().values()) {
            context.write(NullWritable.get(), res);
        }
    }
}