本文主要是介绍mapreduce top K实现，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
主题思想：
通过TreeMap数据结构实现在各个Map端预计算top k，再由reduce聚合
public class TopK extends Configured implements Tool {
public static class TopKMapper extends Mapper<Object, Text, NullWritable, LongWritable> {
public static final int K = 100;
private TreeMap<Long, Long> tm = new TreeMap<Long, Long>();
@Override
protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
try {
long k = Integer.parseInt(value.toString().substring(0, 9));
tm.put(k, k);
if (tm.size() > K) {
tm.remove(tm.firstKey());
}
} catch (Exception e) {
context.getCounter("TopK", "errorlog").increment(1);
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
for (Long text : tm.values()) {
context.write(NullWritable.get(), new LongWritable(text));
}
}
}
public static class TopKReducer extends Reducer<NullWritable, LongWritable, NullWritable, LongWritable> {
public static final int K = 100;
private TreeMap<Long, Long> mt = new TreeMap<Long, Long>();
@Override
protected void reduce(NullWritable key, Iterable<LongWritable> values, Context context)
throws IOException, InterruptedException {
for (LongWritable value : values) {
mt.put(value.get(), value.get());
if (mt.size() > K) {
mt.remove(mt.firstKey());
}
}
for (Long val : mt.descendingKeySet()) {
context.write(NullWritable.get(), new LongWritable(val));
}
}
}
@Override
public int run(String[] args) throws Exception {
Configuration conf = getConf();
Job job = new Job(conf, "TopKNum");
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(LongWritable.class);
job.setMapperClass(TopKMapper.class);
job.setReducerClass(TopKReducer.class);
job.setJarByClass(TopK.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
return job.waitForCompletion(true) ? 0 : 1;
}
public static void main(String[] args) throws IOException, InterruptedException {
try {
if (args.length < 2) {
System.err.println("ERROR: Parameter format length ");
System.exit(0);
}
int ret = ToolRunner.run(new TopK(), args);
System.exit(ret);
} catch (Exception e) {
e.printStackTrace();
}
}
}
这篇关于mapreduce top K实现的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
