Mapreduce实例-TopK
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(); } }}
上面是求最大100个，如果求最小的100 个数，改map和reduce中的mt.remove(mt.firstKey());为mt.remove(mt.lastKey()）