MapReduce案例-实现Top N算法
使用MapReduce解决从一组数据中找出Top N项(N>0)。
我们假设所有的输入的key都是唯一的。也就是说，对于给定的一组输入对{(K,V)}，所有的K都是唯一的。
问题描述
现在有如下格式的电影评分数据集hot_movies.csv（这里显示了部分）：
20645098,8.2,小王子 26259677,8.3,垫底辣妹 11808948,7.2,海绵宝宝 26253733,6.4,突然变异 25856265,6.7,烈日迷踪 26274810,6.6,侦探：为了原点 25889465,6.3,抢劫 1972724,7.3,斯坦福监狱实验 6845667,8.0,秘密特工 ......
要求：找出评分最高的5部电影。
实现思路
1、Top N在MySQL中的实现：
SELECT movie_id, movie_score, movie_title FROM hot_movies ORDER BY movie_score DESC LIMIT 5;
2、Top N在Java中的实现：
在Java中实现Top N最简单的方式是使用接口SortedMap<K,V>及其实现类TreeMap<K,V>。SortedMap会对其元素自动排序。
假设我们将元素都保存在topN中，那么需要保持topN的数量为N。 每当加入一个元素时，如果topN.size()>N的话，则从topN中移除第一个元素(最小的那个元素)
3、MapReduce实现Top N方案：
每个Mapper找出本地的Top N列表(N>0)，然后将它传给同一个Reducer。然后这个Reducer从这些Mappers传过来的所有的本地Top N列表中找出最终的Top N项。
Top N算法实现过程如下。输入数据被分为小的数据块，每个数据块对应一个mapper。每个mapper产生一个本地的top 10项，然后发送给reducer。
要参数化top N列表，需要从驱动程序(启动MapReduce job的程序)将N传给map()和reduce()方法(通过使用MapReduce Configuration对象)。 在驱动程序中设置"top.n"参数，然后map()和reduce()方法在各自的setup()方法中读取此参数。
一、创建Java Maven项目
Maven依赖：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>HadoopDemo</groupId>
<artifactId>com.xueai8</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<!--hadoop依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>3.3.1</version>
</dependency>
<!--hdfs文件系统依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>3.3.1</version>
</dependency>
<!--MapReduce相关的依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>3.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-jobclient</artifactId>
<version>3.3.1</version>
</dependency>
<!--junit依赖-->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<!--编译器插件用于编译拓扑-->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<!--指定maven编译的jdk版本和字符集,如果不指定,maven3默认用jdk 1.5 maven2默认用jdk1.3-->
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source> <!-- 源代码使用的JDK版本 -->
<target>1.8</target> <!-- 需要生成的目标class文件的编译版本 -->
<encoding>UTF-8</encoding><!-- 字符集编码 -->
</configuration>
</plugin>
</plugins>
</build>
</project>
TopNMapper.java：
package com.xueai8.topn;
import java.io.IOException;
import java.util.Iterator;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/**
*
* MapReduce 实现Top N 算法
*
* 要求：找出评分最高的5部电影
* hot_movies.csv,字段:
电影ID
豆瓣评分
电影名称
*/
public class TopNMapper extends Mapper<LongWritable, Text, NullWritable, Text> {
private int N = 5; // 默认是 top 5
//private SortedMap<Double, Text> top = new TreeMap<Double, Text>();
private SortedMap<Double, String> top = new TreeMap<>();
@Override
protected void setup(Context context) throws IOException, InterruptedException {
this.N = context.getConfiguration().getInt("N", 5); // 默认是 top 5
}
@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 转换，分词 20645098,8.2,小王子 -> ["20645098","8.2","小王子"]
String[] tokens = value.toString().split(",");
// 提取评分字段。 评分 = tokens[1]
Double score = Double.parseDouble(tokens[1]);
// 将 {评分, 记录} 保存到 SortedMap 中
top.put(score, value.toString());
// 只保存 top N个
if (top.size() > N) {
top.remove(top.firstKey());
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
// 将 SortedMap 中的记录写出
for (String text : top.values()) {
context.write(NullWritable.get(), new Text(text));
}
}
}
TopNReducer.java
package com.xueai8.topn;
import java.io.IOException;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
*
* MapReduce 实现Top N 算法
*/
public class TopNReducer extends Reducer<NullWritable, Text, NullWritable, Text> {
private int N = 5; // 默认
// private SortedMap<Double, Text> top = new TreeMap<Double, Text>();
private TreeMap<Double, String> top = new TreeMap<>();
@Override
protected void setup(Context context) throws IOException, InterruptedException {
this.N = context.getConfiguration().getInt("N", 5); // 默认是top 5
}
@Override
public void reduce(NullWritable key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
for (Text value : values) {
// 转换，分词 20645098,8.2,小王子 -> ["20645098","8.2","小王子"]
String[] tokens = value.toString().trim().split(",");
// 提取评分字段。 评分 = tokens[1]
Double score = Double.parseDouble(tokens[1]);
// 将 {评分, 记录} 保存到 SortedMap 中
top.put(score, value.toString());
// 只保存 top N
if (top.size() > N) {
top.remove(top.firstKey());
}
}
// 依次输出 top n 项（降序）
for (String text : top.descendingMap().values()) {
context.write(NullWritable.get(), new Text(text));
}
}
}
TopNDriver.java：
package com.xueai8.topn;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
*
* Top N问题：找出评分最高的 5 部电影
*
* 假设所有的 key（评分）是唯一的
*/
public class TopNDriver extends Configured implements Tool {
public static void main(String[] args) throws Exception {
if (args.length != 2) {
System.out.println("用法：TopNDriver <input> <output>");
System.exit(1);
}
int returnStatus = ToolRunner.run(new TopNDriver(), args);
System.exit(returnStatus);
}
public int run(String[] args) throws Exception {
Job job = Job.getInstance(getConf(), "TopN Demo");
job.setJarByClass(TopNDriver.class);
// int N = Integer.parseInt(args[0]); // top N
// job.getConfiguration().setInt("N", N); // 将命令行传的参数N设置到conf中
job.setMapperClass(TopNMapper.class);
job.setMapOutputKeyClass(NullWritable.class);
job.setMapOutputValueClass(Text.class);
job.setReducerClass(TopNReducer.class);
job.setNumReduceTasks(1);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(Text.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
boolean status = job.waitForCompletion(true);
return status ? 0 : 1;
}
}
二、配置log4j
在src/main/resources目录下新增log4j的配置文件log4j.properties，内容如下：
log4j.rootLogger = info,stdout
### 输出信息到控制抬 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n
三、项目打包
打开IDEA下方的终端窗口terminal，执行"mvn clean package"打包命令，如下图所示：
如果一切正常，会提示打jar包成功。如下图所示：
这时查看项目结构，会看到多了一个target目录，打好的jar包就位于此目录下。如下图所示：
四、项目部署
请按以下步骤执行。
1、启动HDFS集群和YARN集群。在Linux终端窗口中，执行如下的脚本：
$ start-dfs.sh
$ start-yarn.sh
查看进程是否启动，集群运行是否正常。在Linux终端窗口中，执行如下的命令：
$ jps
这时应该能看到有如下5个进程正在运行，说明集群运行正常：
5542 NodeManager
5191 SecondaryNameNode
4857 NameNode
5418 ResourceManager
4975 DataNode
2、将数据文件hot_movies.csv上传到HDFS的/data/mr/目录下。
$ hdfs dfs -mkdir -p /data/mr $ hdfs dfs -put hot_movies.csv /data/mr/ $ hdfs dfs -ls /data/mr/
3、提交作业到Hadoop集群上运行。（如果jar包在Windows下，请先拷贝到Linux中。）
在终端窗口中，执行如下的作业提交命令：
$ hadoop jar com.xueai8-1.0-SNAPSHOT.jar com.xueai8.topn.TopNDriver /data/mr /data/mr-output
4、查看输出结果。
在终端窗口中，执行如下的HDFS命令，查看输出结果：
$ hdfs dfs -ls /data/mr-output
可以看到如下的输出结果：
19897541,9.0,机动战士高达 THE ORIGIN I 青瞳的卡斯巴尔 26393561,8.8,小萝莉的猴神大叔 25955491,8.6,罪恶之家 3592854,8.5,疯狂的麦克斯4：狂暴之路 23761370,8.4,速度与激情7
