MapReduce案例-二次排序算法_示例1
MapReduce框架基于key-value对的key对Reduce输入数据排序，还基于该key对数据分组。Hadoop为排序后的key中的每个唯一key调用reduce函数，排序后的key带有属于该key的值列表作为第二个参数。
但是，每个key的值列表是无序的。有许多场景需要每个key的Reduce输入值列表基于一定的条件排序。 例如，对于一个给定的key，找出值的最大值或最小值，而不需要迭代整个值列表。再例如，优化Reduce端的join，识别重复的数据产出等。
问题描述
在sample_input.txt文件中存储有日期和温度数据，数据格式：“年,月,日,温度”：
2000,12,04,10 2000,11,01,20 2000,12,02,-20 2000,11,07,30 2000,11,24,-40 2012,12,21,30 2012,12,22,-20 2012,12,23,60 2012,12,24,70 2012,12,25,10 2013,01,22,80 2013,01,23,90 2013,01,24,70 2013,01,20,-10
要求对数据进行排序，输出数据格式："年-月:温度1,温度2,温度3, ...。其中温度1<=温度2<=温度3<=...。
2012-01: 5, 10, 35, 45, ... 2001-11: 40, 46, 47, 48, ... 2005-08: 38, 50, 52, 70, ... ......
实现思路
我们可以使用Hadoop框架中一个叫做”二次排序”的机制来排序Reduce输入值。其原理如下。
MapReduce工作流程：
数据流图如下：
可以看出，map输出数据格式为：<<年月-温度>,温度>
实现思路：
要实现二次排序特性，我们必须告诉MapReduce/Hadoop框架：
- 怎样排序reducer keys？
- 怎样分区(partition)传递给reducers的keys(自定义partitioner)？
- 怎样分组(group)到达每个reducer的数据？
Map端处理：
(1)复合键设计
使用Value-to-Key转换设计模式：形成一个复合中间键-(K,V1)。 其中v1是次关键字。这里K被称为自然键。 要将一个value(例如，V1)注入到一个reducer key内，简单地创建一个复合键即可。 也就是说，将原始数据的key值和其对应的数据组合成一个新的key值，然后新的key值对应的value还是原始数据中的value。 那么我们只需要对组合键（新key值）进行排序就OK了。
本例中的复合键设计如下：
(2)分区器设计
然后我们需要自定义一个分区处理器，因为我们的目标不是想将新key值相同的记录传到一个reduce中，而是想将新key值中第一个字段相同的记录放到同一个reduce中进行分组合并，所以我们需要根据新key值的第一个字段来自定义一个分区处理器。
分区操作完成之后，再调用自己的自定义排序器对新的key值进行排序。
Reduce端处理：
经过Shuffle处理之后，数据传输到Reducer端了。 在Reducer端按照组合键的第一个字段进行分组，并且每处理完一次分组之后就会调用一次reduce函数来对这个分组进行处理和输出。 所以这里需要自定义一个分组比较器。
一、创建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>
DateTemperaturePair.java：
首先，定义一个可比较的key数据类型，用作复合键类型。
package com.xueai8.secondarysort;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
/**
* 定义Hadoop Key和Value的数据类型，实现Writable和WritableComparable接口
*/
public class DateTemperaturePair implements Writable, WritableComparable<DateTemperaturePair> {
private Text yearMonth = new Text(); // natural key
private Text day = new Text();
private IntWritable temperature = new IntWritable(); // secondary key
public DateTemperaturePair() {
}
public DateTemperaturePair(String yearMonth, String day, int temperature) {
this.yearMonth.set(yearMonth);
this.day.set(day);
this.temperature.set(temperature);
}
public Text getYearMonthDay(){
return new Text(this.yearMonth.toString() + this.day.toString());
}
public Text getYearMonth() {
return yearMonth;
}
public void setYearMonth(String yearMonth) {
this.yearMonth.set(yearMonth);
}
public Text getDay() {
return day;
}
public void setDay(String day) {
this.day.set(day);
}
public IntWritable getTemperature() {
return temperature;
}
public void setTemperature(int temperature) {
this.temperature.set(temperature);
}
public static DateTemperaturePair read(DataInput in) throws IOException{
DateTemperaturePair pair = new DateTemperaturePair();
pair.readFields(in);
return pair;
}
@Override
public void readFields(DataInput in) throws IOException {
this.yearMonth.readFields(in);
this.day.readFields(in);
this.temperature.readFields(in);
}
@Override
public void write(DataOutput out) throws IOException {
this.yearMonth.write(out);
this.day.write(out);
temperature.write(out);
}
// 这个比较器控制键的排序顺序：
// 1）先按“年-月”进行比较
// 2）年月相同时，再按温度进行比较
@Override
public int compareTo(DateTemperaturePair pair) {
int compareValue = this.yearMonth.compareTo(pair.getYearMonth());
if(compareValue == 0){
compareValue = temperature.compareTo(pair.getTemperature());
}
// return compareValue; // 升序
return -1*compareValue; // 降序
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + ((yearMonth == null) ? 0 : yearMonth.hashCode());
result = prime * result + ((temperature == null) ? 0 : temperature.hashCode());
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj){
return true;
}
if (obj == null){
return false;
}
if (getClass() != obj.getClass()){
return false;
}
DateTemperaturePair other = (DateTemperaturePair) obj;
if (this.temperature == null) {
if (other.temperature != null){
return false;
}
} else if (!this.temperature.equals(other.temperature)){
return false;
}
if (yearMonth == null) {
if (other.yearMonth != null){
return false;
}
} else if (!yearMonth.equals(other.yearMonth)){
return false;
}
return true;
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("DateTemperaturePair{yearMonth=");
builder.append(yearMonth);
builder.append(",day=");
builder.append(day);
builder.append(",temperature=");
builder.append(temperature);
builder.append("}");
return builder.toString();
}
}
SecondarySortMapper.java
Mapper类。在这里组装复合键。
package com.xueai8.secondarysort;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class SecondarySortMapper extends Mapper<LongWritable,Text,DateTemperaturePair,Text>{
private DateTemperaturePair pair = new DateTemperaturePair(); // reducer key，复合键
private Text theTemperature = new Text(); // reducer value
/**
* @param key 由Hadoop自动生成(这里我们忽略它)
* @param value 读取的一行文本，格式为："YYYY,MM,DD,temperature"
*/
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 先对一行输入文本进行分词
String line = value.toString(); // "2000,12,04,10"
String[] tokens = line.split(","); // ["2000","12","04","10"]
// YYYY = tokens[0]
// MM = tokens[1]
// DD = tokens[2]
// temperature = tokens[3]
// 提取字段，并组合
String yearMonth = tokens[0] + "-" + tokens[1]; // "2000-12"
String day = tokens[2]; // "04"
int temperature = Integer.parseInt(tokens[3]); // 10
// 准备reducer key（复合键）
pair.setYearMonth(yearMonth);
pair.setDay(day);
pair.setTemperature(temperature);
// 准备reducer value
theTemperature.set(tokens[3]);
// 发送到reducer
context.write(pair, theTemperature); // => <Object,int>
}
}
DateTemperaturePartitioner.java：
DateTemperaturePartitioner是一个自定义的分区器类，它按自然key（使用yearMonth）来分区数据。 在Hadoop中，分区阶段发生在map()阶段之后，reduce()阶段之前。 这个分区器(Partitioner)确保所有具有相同key的数据都会被发送到相同的reducer。
package com.xueai8.secondarysort;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
/**
* DateTemperaturePartitioner是一个自定义的分区器类，它按自然key（使用yearMonth）来分区数据。
* 如果没有自定义分区程序，Hadoop将基于散列代码对映射数据进行分区（哈希分区）。
*
*/
public class DateTemperaturePartitioner extends Partitioner<DateTemperaturePair, Text> {
@Override
public int getPartition(DateTemperaturePair pair, Text text, int numberOfPartitions) {
// 确保分区是非负的
return Math.abs(pair.getYearMonth().hashCode() % numberOfPartitions);
}
}
DateTemperatureGroupingComparator.java：
在调用Reducer的reduce函数之前，Reducer先通过DateTemperatureGroupingComparator进行组内排序。
package com.xueai8.secondarysort;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
/**
* DateTemperatureGroupingComparator类使我们能够比较两个DateTemperaturePair对象。
* 这个类用于排序目的。-用于分区器分组
*
*/
public class DateTemperatureGroupingComparator extends WritableComparator {
public DateTemperatureGroupingComparator(){
super(DateTemperaturePair.class,true);
}
// 这个比较器控制哪些键被组合成一个对reduce()方法的调用
/**
* @param wc1 一个WritableComparable对象，代表一个DateTemperaturePair
* @param wc2 一个WritableComparable对象，代表一个DateTemperaturePair
* @return 0,1,或 -1 (取决于两个DateTemperaturePair对象的比较)
*/
@Override
public int compare(WritableComparable wc1, WritableComparable wc2) {
DateTemperaturePair pair1 = (DateTemperaturePair)wc1;
DateTemperaturePair pair2 = (DateTemperaturePair)wc2;
return pair1.getYearMonth().compareTo(pair2.getYearMonth());
}
}
SecondarySortReducer.java：
在调用reduce之前，已经由DateTemperatureGroupingComparator进行了组内排序。 在定义key数据类型时，已经指定了比较器：先按年月排序，相同年月按温度排序。
package com.xueai8.secondarysort;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
// 自定义的分区器会把相同yearMonth的Object发送到同一个Reducer中
public class SecondarySortReducer extends Reducer<DateTemperaturePair,Text,Text,Text>{
// 在调用reduce之前，先通过DateTemperatureGroupingComparator组内排序
// 输入<Object, [temp1,temp2,temp3,...]>
@Override
protected void reduce(DateTemperaturePair key, Iterable<Text> values,Context context)
throws IOException, InterruptedException {
StringBuilder builder = new StringBuilder();
for(Text value : values){
builder.append(value.toString());
builder.append(",");
}
// 写出
context.write(key.getYearMonth(), new Text(builder.toString()));
}
}
SecondarySortDriver.java：
驱动程序类，用于将作业提交到Hadoop上执行。
package com.xueai8.secondarysort;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
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.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;
/**
* SecondarySortDriver驱动程序类，用于将作业提交到Hadoop上执行
*
*/
public class SecondarySortDriver extends Configured implements Tool{
private static Logger theLogger = Logger.getLogger(SecondarySortDriver.class);
public static void main(String[] args) throws Exception{
if(args.length != 2){
theLogger.warn("语法：SecondarySortDirver <input-dir> <output-dir>");
throw new IllegalArgumentException("SecondarySortDriver <input-dir> <output-dir>");
}
int returnStatus = ToolRunner.run(new SecondarySortDriver(), args);
theLogger.info("returnStatus=" + returnStatus);
System.exit(returnStatus);
}
@Override
public int run(String[] args) throws Exception {
Configuration conf = getConf();
Job job = Job.getInstance(conf, "SecondarySortDriver");
job.setJarByClass(SecondarySortDriver.class);
// args[0] = input directory
// args[1] = output directory
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setMapOutputKeyClass(DateTemperaturePair.class);
job.setMapOutputValueClass(Text.class);
job.setMapperClass(SecondarySortMapper.class);
job.setReducerClass(SecondarySortReducer.class);
/*** 注意这里：分别设置自定义的分区程序和组内比较器 ***/
job.setPartitionerClass(DateTemperaturePartitioner.class);
job.setGroupingComparatorClass(DateTemperatureGroupingComparator.class);
// 提交作业
boolean status = job.waitForCompletion(true);
theLogger.info("run(): status=" + status);
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、将数据文件sample_input.txt上传到HDFS的/data/mr/目录下。
$ hdfs dfs -mkdir -p /data/mr $ hdfs dfs -put sample_input.txt /data/mr/ $ hdfs dfs -ls /data/mr/
3、提交作业到Hadoop集群上运行。（如果jar包在Windows下，请先拷贝到Linux中。）
在终端窗口中，执行如下的作业提交命令：
$ hadoop jar com.xueai8-1.0-SNAPSHOT.jar com.xueai8.secondarysort.SecondarySortDriver /data/mr /data/mr-output
4、查看输出结果。
在终端窗口中，执行如下的HDFS命令，查看输出结果：
$ hdfs dfs -ls /data/mr-output
可以看到如下的输出结果：
2013-01 90,80,70,-10, 2012-12 70,60,30,10,-20, 2000-12 10,-20, 2000-11 30,20,-40,
