package cn.mr.topn;
import java.util.TreeMap;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class TopNMapper extends Mapper {
private TreeMap repToRecordMap = new TreeMap();
//
//
@Override
public void map(LongWritable key, Text value, Context context) {
String line = value.toString();
String[] nums = line.split(” “);
for (String num : nums) {
repToRecordMap.put(Integer.parseInt(num), ” “);
if (repToRecordMap.size() > 5) {
repToRecordMap.remove(repToRecordMap.firstKey());
}
}
}
@Override
protected void cleanup(Context context) {
for (Integer i : repToRecordMap.keySet()) {
try {
context.write(NullWritable.get(), new IntWritable(i));
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
package cn.mr.topn;
import java.io.IOException;
import java.util.Comparator;
import java.util.TreeMap;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
public class TopNReducer extends Reducer {
private TreeMap repToRecordMap = new TreeMap(new Comparator() {
//返回一個(gè)基本類(lèi)型的整型,誰(shuí)大誰(shuí)排后面.
//返回負(fù)數(shù)表示：o1 小于o2
//返回0表示:表示：o1和o2相等
//返回正數(shù)表示：o1大于o2。
public int compare(Integer a, Integer b) {
return b – a;
}
});
public void reduce(NullWritable key, Iterable values, Context context)
throws IOException, InterruptedException {
for (IntWritable value : values) {
repToRecordMap.put(value.get(), ” “);
if (repToRecordMap.size() > 5) {
repToRecordMap.remove(repToRecordMap.firstKey());
}
}
for (Integer i : repToRecordMap.keySet()) {
context.write(NullWritable.get(), new IntWritable(i));
}
}
}
package cn.mr.topn;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class TopNRunner {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(TopNRunner.class);
job.setMapperClass(TopNMapper.class);
job.setReducerClass(TopNReducer.class);
job.setNumReduceTasks(1);
job.setMapOutputKeyClass(NullWritable.class);// map階段的輸出的key
job.setMapOutputValueClass(IntWritable.class);// map階段的輸出的value
job.setOutputKeyClass(NullWritable.class);// reduce階段的輸出的key
job.setOutputValueClass(IntWritable.class);// reduce階段的輸出的value
FileInputFormat.setInputPaths(job, new Path(“D:topNinput”));
FileOutputFormat.setOutputPath(job, new Path(“D:topNoutput”));
boolean res = job.waitForCompletion(true);
System.exit(res ? 0 : 1);
}
}
