<>前言

在很多业务场景下,需要对原始的数据读取分析后,将输出的结果按照指定的业务字段进行排序输出,方便上层应用对结果数据进行展示或使用,减少二次排序的成本

在hadoop的MapReduce中,提供了对于客户端的自定义排序的功能相关API

<>MapReduce排序

* 默认情况下,MapTask 和ReduceTask均会对数据按照key进行排序
* 默认的排序按照字典序,且实现排序的方法是快排
<>MapReduce排序分类

1、部分排序

MapReduce根据输入记录的键值对数据集总体排序,确保输出的文件内部数据有序

2、全排序

最终的输出结果只有一个文件,且内部有序,实现方式是只设置一个ReduceTask,但是这种做法在处理的某个文件特别大的时候,效率会非常低,这也就丧失了MapReduce提供的并行处理任务的能力

3、辅助排序

在Reduce端对key进行分组,比如说,在接收的key为bean对象的时候,想让一个或多个字段相同的key进入到同一个reduce方法时,可以采用分组排序

4、二次排序

在自定义排序中,compareto的判断条件为两个或者多个时即为二次排序

<>自定义排序案例

还记得在序列化一篇中,那个针对手机号的峰值流量和峰谷流量的例子吧,我们直接以该案例的输出结果为输入数据,对这个结果文件中按照总流量进行排序

期望输出数据的格式如:

1、自定义一个Bean对象,实现WritableComparable 接口

实现该接口后,重写compareTo方法,需要排序的字段逻辑就在compareTo中编写
import org.apache.hadoop.io.WritableComparable; import java.io.DataInput;
import java.io.DataOutput; import java.io.IOException; public class
PhoneSortBean implements WritableComparable<PhoneSortBean> { //峰值流量 private
long upFlow; //低谷流量 private long downFlow; //总流量 private long sumFlow;
@Override public int compareTo(PhoneSortBean o) { if (this.sumFlow > o.sumFlow)
{ return -1; }else if(this.sumFlow < o.sumFlow){ return 1; }else { return 0; }
} //提供无参构造 public PhoneSortBean() { } //提供三个参数的getter和setter方法 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; } public void setSumFlow() { this.sumFlow = this.upFlow
+ this.downFlow; } //实现序列化和反序列化方法,注意顺序一定要保持一致 @Override public void
write(DataOutput dataOutput) throws IOException { dataOutput.writeLong(upFlow);
dataOutput.writeLong(downFlow); dataOutput.writeLong(sumFlow); } @Override
public void readFields(DataInput dataInput) throws IOException { this.upFlow =
dataInput.readLong(); this.downFlow = dataInput.readLong(); this.sumFlow =
dataInput.readLong(); } //重写ToString方法 @Override public String toString() {
return upFlow + "\t" + downFlow + "\t" + sumFlow; } }
2、自定义Mapper

设想一下,既然数据能排序,Map阶段输出的key应该为自定义的可比较的对象,即为上面的这个bean,value为手机号
import org.apache.commons.lang3.StringUtils; import
org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import
org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; import
java.util.LinkedList; public class SortPhoneMapper extends Mapper<LongWritable,
Text, PhoneSortBean,Text> { private Text outV = new Text(); private
PhoneSortBean outK = new PhoneSortBean(); @Override protected void
map(LongWritable key, Text value, Context context) throws IOException,
InterruptedException { String line = value.toString(); //分割数据 String[] splits =
line.split("\t"); LinkedList<String> linkedList = new LinkedList<>();
for(String str:splits){ if(StringUtils.isNotEmpty(str)){
linkedList.add(str.trim()); } } //抓取需要的数据:手机号,上行流量,下行流量 String phone =
linkedList.get(0); String max = linkedList.get(1); String mine =
linkedList.get(2); //封装outK outV outV.set(phone);
outK.setUpFlow(Long.parseLong(max)); outK.setDownFlow(Long.parseLong(mine));
outK.setSumFlow(); //写出outK outV context.write(outK, outV); } }
3、自定义Reducer

Reduce阶段的输出结果仍然以手机号为key,而value为排序后的自定义的bean
import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException; public class SortPhoneReducer extends
Reducer<PhoneSortBean,Text , Text, PhoneSortBean> { @Override protected void
reduce(PhoneSortBean key, Iterable<Text> values, Context context) throws
IOException, InterruptedException { for (Text value : values) {
context.write(value,key); } } }
4、自定义Driver类
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; public class
SortPhoneJob { public static void main(String[] args) throws Exception { //1
获取job对象 Configuration conf = new Configuration(); Job job =
Job.getInstance(conf); //2 关联本Driver类 job.setJarByClass(SortPhoneJob.class);
//3 设置Map端输出KV类型 job.setReducerClass(SortPhoneReducer.class);
job.setMapperClass(SortPhoneMapper.class); //4 关联Mapper和Reducer
job.setMapOutputKeyClass(PhoneSortBean.class);
job.setMapOutputValueClass(Text.class); //5 设置程序最终输出的KV类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(PhoneSortBean.class); //6 设置程序的输入输出路径 String inPath =
"F:\\网盘\\csv\\phone_out_bean.txt"; String outPath =
"F:\\网盘\\csv\\phone_out_sort"; FileInputFormat.setInputPaths(job, new
Path(inPath)); FileOutputFormat.setOutputPath(job, new Path(outPath)); //7
提交Job boolean b = job.waitForCompletion(true); System.exit(b ? 0 : 1); } }
运行上面的程序,观察输出结果,可以看到,总流量按照从大到小的顺序进行了排序

可以看到,最后的3行数据中,总流量相同,如果这时候又提出一个需求,当总流量相同时,再按照峰值流量进行排序,该怎么做呢?

其实只需要在自定义的bean中的compareto方法里面继续添加排序逻辑即可
public int compareTo(PhoneSortBean o) { if (this.sumFlow > o.sumFlow) { return
-1; }else if(this.sumFlow < o.sumFlow){ return 1; }else {
//如果总流量相同的情况下,再按照峰值流量排序 if(this.upFlow > o.upFlow){ return -1; }else
if(this.upFlow < o.upFlow){ return 1; }else { return 0; } } }
<>分区内排序案例

业务需求,上面的案例中,我们进一步提出新的需求,针对不同的手机号最终写到不同的文件中,那么在上面的基础上,还需要结合自定义分区的逻辑

需要改造的包括2点:

* 添加一个自定义分区器,按照业务规则指定分区号
* 改造Driver类,添加自定义分区器,设置MapReduceTask任务个数
1、添加自定义分区
public class MyPartioner extends Partitioner<MyPhoneBean, Text> { @Override
public int getPartition(MyPhoneBean myPhoneBean, Text text, int partion) {
String phone = text.toString(); if(phone.startsWith("135")){ return 0; }else
if(phone.startsWith("136")){ return 1; }else if(phone.startsWith("137")){
return 2; }else { return 3; } } }
2、改造Driver类

其他的逻辑和上面的保持一致即可
public class MyDriver { public static void main(String[] args) throws
Exception { //1 获取job对象 Configuration conf = new Configuration(); Job job =
Job.getInstance(conf); //2 关联本Driver类 job.setJarByClass(MyDriver.class); //3
设置Map端输出KV类型 job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class); //4 关联Mapper和Reducer
job.setMapOutputKeyClass(MyPhoneBean.class);
job.setMapOutputValueClass(Text.class); //5 设置程序最终输出的KV类型
job.setOutputKeyClass(Text.class); job.setOutputValueClass(MyPhoneBean.class);
//6、设置输出文件为2个 job.setNumReduceTasks(4);
job.setPartitionerClass(MyPartioner.class); //7、 设置程序的输入输出路径 String inPath =
"F:\\网盘\\csv\\phone_out_bean.txt"; String outPath =
"F:\\网盘\\csv\\phone_out_sort"; FileInputFormat.setInputPaths(job, new
Path(inPath)); FileOutputFormat.setOutputPath(job, new Path(outPath)); //7
提交Job boolean b = job.waitForCompletion(true); System.exit(b ? 0 : 1); } }
运行上面的程序,然后随机打开其中的两个文件检查下是否满足上面的需求,可以看到,文件最终输出到4个分区文件下,并且每个分区文件内的总流量也是按照从高到低的顺序

技术
©2019-2020 Toolsou All rights reserved,
TypeScript:函数类型接口8道大厂指针笔试题让你秒杀指针!!!MySQL 日期时间加减mysql 查询条件之外的数据_mysql 查询符合条件的数据查linux的操作系统版本,如何查看Linux操作系统版本?将String类型转换成Map数据类型使用uuid做MySQL主键,被老板,爆怼一顿C语言中的字符串函数和字符函数linux服务器中毒排查--基础篇C# ASCII码字符转换