实例描述

• 对数据文件中的数据取 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

Top N

  Top-N 分析法是指从研究对象中得到所需的 N 个数据，并对这 N 个数据进行重点分析的方法。那么该如何利用 MapReduce 来解决在海量数据中求 Top N 个数。

设计思路

  要找出 top N, 核心是能够想到 the number of reduce tasks is one。因为一个 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 执行完之后会执行的一个函数。

• Reducer 过程

  • 只有一个 reducer，就是对 mapper 输出的数据进行再一次汇总，选出其中的 top n，即可达到我们的目的。 注意的是， Treemap 默认是正序排列数据，要想满足求取 top n 倒序最大的 n 个，需要实现自己的 Comparator（） 方法。

测试数据

aaa.txt

10 3 8 7 6 5 1 2 9 4
11 12 17 14 15 20
19 18 13 16

程序代码

package cn.mn1024.mapreduce.dedup;

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 功能：TopN
 * @author GuangLing_Lin
 */
@SuppressWarnings("all")
public class TopNRunner {

    private static class TopNMapper extends Mapper<LongWritable, Text, NullWritable, IntWritable> {

        private static 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();
                }
            }
        }

    }

    private static class TopNReducer extends Reducer<NullWritable, IntWritable, NullWritable, IntWritable> {
        private static 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;
            }
        });

        @Override
        protected 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));
            }
        }
    }

    public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
        // 创建本次mr程序的job实例
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        
        // 指定本次job运行的主类
        job.setJarByClass(TopNRunner.class);
        
        // 指定本次job的具体Mapper、Reducer的实现类
        job.setMapperClass(TopNMapper.class);
        job.setReducerClass(TopNReducer.class);
        
        job.setNumReduceTasks(1);
        
        // 指定本次job map阶段输出数据类型
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        // 指定本次job reduce阶段输出数据类型 也就是整个mr任务的最终输出数据类型
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(IntWritable.class);
        
        // 指定本次job待处理数据的目录和程序执行完毕结果存放的目录
        FileInputFormat.setInputPaths(job, new Path("F://topN//input"));
        FileOutputFormat.setOutputPath(job, new Path("F://topN//output"));
        
        // 提交本次job
        boolean res = job.waitForCompletion(true);
        System.exit(res ? 0 : 1);
    }

}

测试结果

20
19
18
17
16

每一个成功的背后都有无数个无人知晓的黑夜。

因为

夜晚，是超越对手的最佳时机。

===================== 码农1024 =====================#蔺光岭#