MapReduce 案例之Top N
1. Top N
Top-N 分析法是指从研究对象中得到所需的 N 个数据，并对这 N 个数据进行重点分析的方法。
2. 实例描述
对数据文件中的数据取最大 top-n。数据文件中的每个都是一个数据。
原始输入数据为：
10 3 8 7 6 5 1 2 9 4
11 12 17 14 15 20
19 18 13 16
输出结果为（最大的前 5 个）：
20
19
18
17
16
3. 设计思路
要找出 top N, 核心是能够想到 reduce k Task 个数 一定只有一个。
因为一个 map task 就是一个进程,有几个 map task 就有几个中间文件，有几个 reduce task 就有几个最终输出文件。我们要找的 top N 是指的全局的前 N 条数据，那么不管中间有几个 map, reduce 最终只能有一个 reduce 来汇总数据，输出 top N。
- Mapper 过程
使用默认的 mapper 数据，一个 input split（输入分片）由一个 mapper 来处理。
在每一个 map task 中，我们找到这个 input split 的前 n 个记录。这里我们用 TreeMap这个数据结构来保存 top n 的数据，TreeMap 默认按键的自然顺序升序进行排序。下一步，我们来加入新记录到 TreeMap 中去。在 map 中，我们对每一条记录都尝试去更新 TreeMap，最后我们得到的就是这个分片中的 local top n 的 n 个值。
以往的 mapper 中，我们都是处理一条数据之后就 context.write 一次。而在这里是把所有这个 input split 的数据处理完之后再进行写入。所以，我们可以把这个 context.write放在 cleanup 里执行。cleanup 就是整个 mapper task 执行完之后会执行的一个函数。
TreeMap 是一个有序的 key-value 集合，默认会根据其键的自然顺序进行排序，也可根据创建映射时提供的 Comparator 进行排序。其 firstKey()方法用于返回当前这个集合第一个(最低)键。 - Reducer 过程
只有一个 reducer，就是对 mapper 输出的数据进行再一次汇总，选出其中的 top n，即可达到我们的目的。注意的是，Treemap 默认是正序排列数据，要想满足求取 top n 倒序最大的 n 个，需要实现自己的 Comparator（）方法。
4. 程序代码
- pom文件参考MapReduce 案例之倒排索引
http://blog.csdn.net/qq_41455420/article/details/79294951 - Map程序
package cn.itcast.hadoop.topn; import java.util.StringTokenizer; 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<LongWritable, Text, NullWritable, IntWritable> { private TreeMap<Integer, String> repToRecordMap = new TreeMap<Integer, String>(); @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(); } } } }
- reduce程序
package cn.itcast.hadoop.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<NullWritable, IntWritable, NullWritable, IntWritable> { private TreeMap<Integer, String> repToRecordMap = new TreeMap<Integer, String>(new Comparator<Integer>() { /* * int compare(Object o1, Object o2) 返回一个基本类型的整型， * 返回负数表示：o1 小于o2， * 返回0 表示：o1和o2相等， * 返回正数表示：o1大于o2。 * 谁大谁排后面 */ public int compare(Integer a, Integer b) { return b - a; } }); public void reduce(NullWritable key, Iterable<IntWritable> 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.itcast.hadoop.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:\\topN\\input")); FileOutputFormat.setOutputPath(job, new Path("D:\\topN\\output")); boolean res = job.waitForCompletion(true); System.exit(res ? 0 : 1); } }
