MapReduce程序之TopN问题（排行榜问题）
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了MapReduce程序之TopN问题（排行榜问题）相关的知识，希望对你有一定的参考价值。
[toc]MapReduce程序之TopN问题（排行榜问题）
需求
有下面的文本文件：
[email protected]:~/data/input/topn$ cat senventeen_a.txt
1,9819,100,121
2,8918,2000,111
3,2813,1234,22
4,9100,10,1101
5,3210,490,111
6,1298,28,1211
7,1010,281,90
8,1818,9000,20
[email protected]:~/data/input/topn$ cat senventeen_b.txt
100,3333,10,100
101,9321,1000,293
102,3881,701,20
103,6791,910,30
104,8888,11,39以逗号作为分隔符，每一列分别为orderid,userid,payment,productid，现在需要按照payment从大到小求出TopN，比如top10，其输出结果应该如下：
1 9000
2 2000
3 1234
4 1000
5 910
6 701
7 490
8 281
9 100
10 28此外，TopN中的N应该是动态的，由输入的参数来决定，根据引写一个MapReduce程序来进行处理。
程序思路分析
如下：
Mapper：
/**
* Mapper，因为Block中的每一个split都会交由一个Mapper Task来进行处理，对于TopN问题，可以考虑每一个Mapper Task的输出
* 可以为这个split中的前N个值，最后每个数据到达Reducer的时候，就可以大大减少原来需要比较的数据量，因为在Reducer处理之前
* Map Task已经帮我们把的数据量大大减少了，比如，在MapReduce中，默认情况下一个Block就为一个split，当然这个是可以设置的
* 而一个Block为128M，显然128M能够存储的文本文件也是相当多的，假设现在我的数据有10个Block，即1280MB的数据，如果要求Top10
* 的问题，此时，这些数据需要10个Mapper Task来进行处理，那么在每个Mapper Task中先求出前10个数，最后这10个数再交由Reducer来进行处理
* 也就是说，在我们的这个案例中，Reducer需要处理排序的数有100个，显然经过Map处理之后，Reducer的压力就大大减少了。
* 那么如何实现每个Mapper Task中都只输出10个数呢？这时可以使用一个set来缓存数据，从而达到先缓存10个数的目的，详细可以参考下面的代码。
*/
Reducer：
/**
* Reducer，将Mapper Task输出的数据排序后再输出
* 处理思路与Mapper是类似的
*/
TopN中的N值问题：
// 向conf中传入参数
// 在MapReduce中，因为计算是分散到每个节点上进行的
// 也就是将我们的Maper和Reducer也是分散到每个节点进行的
// 所以不能在TopNJob中设置一个全局变量来对N进行设置（虽然在本地运行时是没有问题的，但在集群运行时会有问题）
// 因此MapReduce提供了在Configuration对象中设置参数的方法
// 通过在Configuration对象中设置某些参数，可以保证每个节点的Mapper和Reducer都能够读取到NMapReduce程序
关于如何处理TopN问题的思路已经在代码注释中有说明，不过需要注意的是，这里使用了前面开发的Job工具类来开发驱动程序。
package com.uplooking.bigdata.mr.topn;
import com.uplooking.bigdata.common.utils.MapReduceJobUtil;
import com.uplooking.bigdata.mr.secondsort.AccessLogWritable;
import com.uplooking.bigdata.mr.secondsort.SecondSortJob;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import java.io.IOException;
import java.util.Comparator;
import java.util.TreeSet;
/**
* MapReduce程序之TopN问题
*/
public class TopNJob {
/**
* 驱动程序，使用Job工具类来生成job
*/
public static void main(String[] args) throws Exception {
if (args == null || args.length < 3) {
System.err.println("Parameter Errors! Usages:<inputpath> <outputpath> <topN>");
System.exit(-1);
}
// 向conf中传入参数
// 在MapReduce中，因为计算是分散到每个节点上进行的
// 也就是将我们的Maper和Reducer也是分散到每个节点进行的
// 所以不能在TopNJob中设置一个全局变量来对N进行设置（虽然在本地运行时是没有问题的，但在集群运行时会有问题）
// 因此MapReduce提供了在Configuration对象中设置参数的方法
// 通过在Configuration对象中设置某些参数，可以保证每个节点的Mapper和Reducer都能够读取到N
Configuration conf = new Configuration();
conf.set("topN", args[2]);
Job job = MapReduceJobUtil.buildJob(conf,
TopNJob.class,
args[0],
TextInputFormat.class,
TopNJobMapper.class,
IntWritable.class,
NullWritable.class,
new Path(args[1]),
TextOutputFormat.class,
TopNReducer.class,
IntWritable.class,
IntWritable.class);
// ReduceTask必须设置为1
job.setNumReduceTasks(1);
job.waitForCompletion(true);
}
/**
* Mapper，因为Block中的每一个split都会交由一个Mapper Task来进行处理，对于TopN问题，可以考虑每一个Mapper Task的输出
* 可以为这个split中的前N个值，最后每个数据到达Reducer的时候，就可以大大减少原来需要比较的数据量，因为在Reducer处理之前
* Map Task已经帮我们把的数据量大大减少了，比如，在MapReduce中，默认情况下一个Block就为一个split，当然这个是可以设置的
* 而一个Block为128M，显然128M能够存储的文本文件也是相当多的，假设现在我的数据有10个Block，即1280MB的数据，如果要求Top10
* 的问题，此时，这些数据需要10个Mapper Task来进行处理，那么在每个Mapper Task中先求出前10个数，最后这10个数再交由Reducer来进行处理
* 也就是说，在我们的这个案例中，Reducer需要处理排序的数有100个，显然经过Map处理之后，Reducer的压力就大大减少了。
* 那么如何实现每个Mapper Task中都只输出10个数呢？这时可以使用一个set来缓存数据，从而达到先缓存10个数的目的，详细可以参考下面的代码。
*/
public static class TopNJobMapper extends Mapper<LongWritable, Text, IntWritable, NullWritable> {
TreeSet<Integer> cachedTopN = null;
Integer N = null;
/**
* 每个Mapper Task执行前都会先执行setup函数
* map函数是每行执行一次
*/
@Override
protected void setup(Context context) throws IOException, InterruptedException {
// TreeSet定义的排序规则为倒序，后面做数据的处理时只需要pollLast最后一个即可将
// TreeSet中较小的数去掉
cachedTopN = new TreeSet<Integer>(new Comparator<Integer>() {
@Override
public int compare(Integer o1, Integer o2) {
int ret = 0;
if (o1 > o2) {
ret = -1;
} else if (o1 < o2) {
ret = 1;
}
return ret;
}
});
// 拿到传入参数时的topN中的N值
N = Integer.valueOf(context.getConfiguration().get("topN"));
}
/**
* 将split中前N个数筛选出来
*/
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 解析每一行
String[] fields = value.toString().split(",");
if (fields == null || fields.length < 3) {
return;
}
// 转换payment为数字，如果出现异常，终止当前map函数的执行
Integer payment = null;
try {
payment = Integer.valueOf(fields[2]);
} catch (NumberFormatException e) {
e.printStackTrace();
return;
}
// 将数字写入到TreeSet当中
cachedTopN.add(payment);
// 判断cachedTopN中的元素个数是否已经达到N个，如果已经达到N个，则去掉最后一个
if (cachedTopN.size() > N) {
cachedTopN.pollLast();
}
}
/**
* 每个Mapper Task执行结束后才会执行cleanup函数
* 将map函数筛选出来的前N个数写入到context中作为输出
* 将
* map函数是每行执行一次
*/
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
for (Integer num : cachedTopN) {
context.write(new IntWritable(num), NullWritable.get());
}
}
}
/**
* Reducer，将Mapper Task输出的数据排序后再输出
* 处理思路与Mapper是类似的
*/
public static class TopNReducer extends Reducer<IntWritable, NullWritable, IntWritable, IntWritable> {
TreeSet<Integer> cachedTopN = null;
Integer N = null;
/**
* 初始化一个TreeSet
*/
@Override
protected void setup(Context context) throws IOException, InterruptedException {
// TreeSet定义的排序规则为倒序，后面做数据的处理时只需要pollLast最后一个即可将
// TreeSet中较小的数去掉
cachedTopN = new TreeSet<Integer>(new Comparator<Integer>() {
@Override
public int compare(Integer o1, Integer o2) {
int ret = 0;
if (o1 > o2) {
ret = -1;
} else if (o1 < o2) {
ret = 1;
}
return ret;
}
});
// 拿到传入参数时的topN中的N值
N = Integer.valueOf(context.getConfiguration().get("topN"));
}
/**
* 筛选Reducer Task中的前10个数
*/
@Override
protected void reduce(IntWritable key, Iterable<NullWritable> values, Context context)
throws IOException, InterruptedException {
cachedTopN.add(Integer.valueOf(key.toString()));
// 判断cachedTopN中的元素个数是否已经达到N个，如果已经达到N个，则去掉最后一个
if (cachedTopN.size() > N) {
cachedTopN.pollLast();
}
}
/**
* 将reduce函数筛选出来的前N个数写入到context中作为输出
*/
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
int index = 1;
for(Integer num : cachedTopN) {
context.write(new IntWritable(index), new IntWritable(num));
index++;
}
}
}
}测试
这里使用本地环境来运行MapReduce程序，输入的参数如下：
/Users/yeyonghao/data/input/topn /Users/yeyonghao/data/output/mr/topn 10也可以将其打包成jar包，然后上传到Hadoop环境中运行。
运行程序后，查看输出结果如下：
[email protected]:~/data/output/mr/topn$ cat part-r-00000
1 9000
2 2000
3 1234
4 1000
5 910
6 701
7 490
8 281
9 100
10 28可以看到，我们的MapReduce程序已经完成了TopN问题的处理，并且其中的N值是动态的，可以根据参数来动态确定。
以上是关于MapReduce程序之TopN问题（排行榜问题）的主要内容，如果未能解决你的问题，请参考以下文章
大数据技术之_05_Hadoop学习_04_MapReduce_Hadoop企业优化(重中之重)+HDFS小文件优化方法+MapReduce扩展案例+倒排索引案例(多job串联)+TopN案例+找博客