上篇博文给大家带来的是分区的介绍以及怎样自定义分区，这次博主为大家带来的是关于排序的博文，希望大家能够喜欢。
一. Shuffle之排序(sort)
今天我们讲的是第六步，sort排序操作。
1.1 排序的简单介绍
排序是MapReduce框架中最重要的操作之一。 MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要否需要。 默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。
对于MapTask，它会将处理的结果暂时放在环形缓冲区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数据溢写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行归并排序。
对于ReduceTask，它从每个MapTask上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中，如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。
1.2 排序的分类
- 1. 部分排序 MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序。
- 2. 全排序 最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行框架。
- 3. 辅助排序：(GroupingComparator分组) 在Reduce端对key进行分组。应用于：在接收的key为bean对象时，想让一个活几个字段相同(全部字段比较不相同)的key进入到同一个reduce方法时，可以采用分组排序。
- 4. 二次排序 在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。
1.3 自定义排序
- 原理分析：
bean对象做为key传输，需要实现
WritableComparable接口重写compareTo方法，就可以实现排序。
@Override
public int compareTo(FlowBean o) {
int result;
// 按照总流量大小，倒序排列
if (sumFlow > bean.getSumFlow()) {
result = -1;
}else if (sumFlow < bean.getSumFlow()) {
result = 1;
}else {
result = 0;
}
return result;
}二. WritableComparable排序案例
案例还是用的上两篇博文的文档。
2.1 需求
代码语言：javascript复制期望输出数据
13509468723 7335 110349 117684
13736230513 2481 24681 27162
13956435636 132 1512 1644
13846544121 264 0 2642.2 需求分析
- 1. 查看需求
- 2. 查看我们需要排序的数据
2.3 编写代码
1. FlowBean对象在在需求1基础上增加了比较功能
代码语言：javascript复制package com.buwenbuhuo.WritableComparable;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
/**
* @author 卜温不火
* @create 2020-04-24 0:33
* com.buwenbuhuo.WritableComparable - the name of the target package where the new class or interface will be created.
* mapreduce0422 - the name of the current project.
*/
public class FlowBean implements WritableComparable<FlowBean> {
private long upFlow;
private long downFlow;
private long sumFlow;
public FlowBean() {
}
@Override
public String toString() {
return upFlow "t" downFlow "t" sumFlow;
}
public void set(long upFlow, long downFlow) {
this.upFlow = upFlow;
this.downFlow = downFlow;
this.sumFlow = upFlow downFlow;
}
public long getUpFlow() {
return upFlow;
}
public void setUpFlow(long upFlow) {
this.upFlow = upFlow;
}
public long getDownFlow() {
return downFlow;
}
public void setDownFlow(long downFlow) {
this.downFlow = downFlow;
}
public long getSumFlow() {
return sumFlow;
}
public void setSumFlow(long sumFlow) {
this.sumFlow = sumFlow;
}
/**
* 序列化方法
* @param out 框架给我们提供的数据出口
* @throws IOException
*/
public void write(DataOutput out) throws IOException {
out.writeLong(upFlow);
out.writeLong(downFlow);
out.writeLong(sumFlow);
}
/**
* 反序列化方法
* @param in 框架提供的数据来源
* @throws IOException
*/
public void readFields(DataInput in) throws IOException {
upFlow = in.readLong();
downFlow = in.readLong();
sumFlow = in.readLong();
}
@Override
public int compareTo(FlowBean o) {
return Long.compare(o.sumFlow, this.sumFlow);
}
}2. 编写Mapper类
代码语言：javascript复制package com.buwenbuhuo.WritableComparable;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/**
* @author 卜温不火
* @create 2020-04-24 0:33
* com.buwenbuhuo.WritableComparable - the name of the target package where the new class or interface will be created.
* mapreduce0422 - the name of the current project.
*/
public class SortMapper extends Mapper<LongWritable, Text, com.buwenbuhuo.WritableComparable.FlowBean, Text> {
private com.buwenbuhuo.WritableComparable.FlowBean flow = new FlowBean();
private Text phone = new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 2 截取
String[] fields = value.toString().split("t");
// 3 封装对象
phone.set(fields[0]);
flow.setUpFlow(Long.parseLong(fields[1]));
flow.setDownFlow(Long.parseLong(fields[2]));
flow.setSumFlow(Long.parseLong(fields[3]));
// 输出
context.write(flow, phone);
}
}3. 编写Reducer类
代码语言：javascript复制package com.buwenbuhuo.WritableComparable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
/**
* @author 卜温不火
* @create 2020-04-24 0:33
* com.buwenbuhuo.WritableComparable - the name of the target package where the new class or interface will be created.
* mapreduce0422 - the name of the current project.
*/
public class SortReducer extends Reducer<com.buwenbuhuo.WritableComparable.FlowBean, Text, Text, com.buwenbuhuo.WritableComparable.FlowBean> {
@Override
protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
// 循环输出，避免总流量相同情况
for (Text value : values) {
context.write(value, key);
}
}
}4. 编写Driver类
代码语言：javascript复制package com.buwenbuhuo.WritableComparable;
import org.apache.hadoop.conf.Configuration;
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 java.io.IOException;
/**
* @author 卜温不火
* @create 2020-04-24 0:33
* com.buwenbuhuo.WritableComparable - the name of the target package where the new class or interface will be created.
* mapreduce0422 - the name of the current project.
*/
public class SortDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 1 获取配置信息，或者job对象实例
Job job = Job.getInstance(new Configuration());
// 2 指定本程序的jar包所在的本地路径job.setJarByClass(com.buwenbuhuo.WritableComparable.SortDriver.class);
// 3 指定本业务job要使用的mapper/Reducer业务类
job.setMapperClass(SortMapper.class);
job.setReducerClass(SortReducer.class);
// 4 指定mapper输出数据的kv类型job.setMapOutputKeyClass(com.buwenbuhuo.WritableComparable.FlowBean.class);
job.setMapOutputValueClass(Text.class);
// 5 指定最终输出的数据的kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBean.class);
// 6 指定job的输入原始文件所在目录
FileInputFormat.setInputPaths(job, new Path("d:\output"));
FileOutputFormat.setOutputPath(job, new Path("d:\output2"));
// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
boolean b = job.waitForCompletion(true);
System.exit(b ? 0 : 1);
}
}2.4 运行及结果
1. 运行
2. 结果
