文章目录
- WritableComparable排序概述
- 第一个案例需求（全排序）
- 代码实现
- 结果分析
- 第二个案例需求（二次排序）
- 问题分析和代码
- 结果分析
- 第三个案例需求（区内排序）
- 需求分析
- 代码实现
- 结果分析
WritableComparable排序概述
排序是MapReducer框架中最重要的操作之一。
MapTask和ReduceTask均会对数据按照key进行排序，该操作属于Hadoop的默认行为，任何应用程序中的数据均会被排序，而不管逻辑上是否需要。
默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。
自定义排序WritableComparable原理分析：
如果要bean对象做为key传输，那么需要实现WritableComparable接口重写compareTo方法，然后就可以实现排序：
@Override
public int compareTo(FlowBean bean) {
int result;
// 按照总流量大小，倒序排列
if (this.sumFlow > bean.getSumFlow()) {
result = -1;
}else if (this.sumFlow < bean.getSumFlow()) {
result = 1;
}else {
result = 0;
}
return result;
}
第一个案例需求（全排序）
根据上一次序列化案例产生的结果（http://t.csdnimg.cn/qogGA）再次对总流量进行倒序排序。
输入数据：
代码实现
package com.atxiaoyu.mapreduce.sort1;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class FlowBean implements WritableComparable<FlowBean> {
private long upFlow; //上行流量
private long downFlow; //下行流量
private long sumFlow; //总流量
//空参构造
public FlowBean() {
}
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 out) throws IOException {
out.writeLong(upFlow);
out.writeLong(downFlow);
out.writeLong(sumFlow);
}
@Override
public void readFields(DataInput in) throws IOException {
this.upFlow=in.readLong();
this.downFlow=in.readLong();
this.sumFlow=in.readLong();
}
@Override
public String toString() {
return upFlow+"\t"+downFlow+"\t"+sumFlow;
}
@Override
public int compareTo(FlowBean o) {
//总流量的倒叙排序
if(this.sumFlow>o.sumFlow){
return -1;
}else if(this.sumFlow<o.sumFlow){
return 1;
}else {
return 0;
}
}
}
package com.atxiaoyu.mapreduce.sort1;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import java.io.IOException;
public class FlowMapper extends Mapper<LongWritable, Text,FlowBean, Text> {
private FlowBean outK=new FlowBean();
private Text outV=new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//获取一行
String line=value.toString();
//切割
String[] split=line.split("\t");
//封装
outV.set(split[0]);
outK.setUpFlow(Long.parseLong(split[1]));
outK.setDownFlow(Long.parseLong(split[2]));
outK.setSumFlow();
//写出
context.write(outK,outV);
}
}
package com.atxiaoyu.mapreduce.sort1;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class FlowReducer extends Reducer<FlowBean,Text,Text, FlowBean> { //总流量相同的，会进入到每一个reducer里面（按照总流量进行排序的）
@Override
protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
for (Text value : values) {
context.write(value,key);
}
}
}
package com.atxiaoyu.mapreduce.sort1;
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;
public class FlowDriver {
public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException {
Configuration conf = new Configuration();
//1 获取job
Job job = Job.getInstance(conf);
//2 设置jar包路径
job.setJarByClass(FlowDriver.class);
// 3 管理mapper和reducer
job.setMapperClass(FlowMapper.class);
job.setReducerClass(FlowReducer.class);
// 4 设置map输出的kv类型
job.setMapOutputKeyClass(FlowBean.class);
job.setMapOutputValueClass(Text.class);
//5 设置最终输出的kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBean.class);
//6 设置输入路径和输出路径
FileInputFormat.setInputPaths(job, new Path("D:\\output"));
FileOutputFormat.setOutputPath(job, new Path("D:\\newOutput"));
//7 提交job
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
结果分析
实现了按照总流量的倒序排序，与我们设想的一致。
第二个案例需求（二次排序）
问题分析和代码
第一个需求我们已经把总流量的倒序排完了，那么会有这么一种情况，如果某几个手机号的总流量相同，但是他们的上行流量和下行流量不相同，这时候我们再加一个功能，就是当总流量相同的时候再按照上行流量的正序排序，我们只需要修改一下compareTo方法里面的逻辑就好了：
@Override
public int compareTo(FlowBean 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;
}
}
}
结果分析
我们修改一下输入文件，着重观察一下这几行数据，都是总流量相同：
然后我们运行看一下输出结果：
与我们设想的一致，实现了当总流量相同的时候再按照上行流量的正序排序。
第三个案例需求（区内排序）
需求分析
基于前一个需求，有一个新的需求，要求按手机号开头前三位实现分区，比如136 137 138 139开头的手机号各在一个文件，其他手机号开头的在一个文件。我们可以通过增加一个自定义分区类来实现。（有关分区的内容在这篇博客中有体现：点击跳转）
代码实现
增加一个自定义分区类ProvincePartitioner2：
package com.atxiaoyu.mapreduce.sort1;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class ProvincePartitioner2 extends Partitioner<FlowBean, Text> {
@Override
public int getPartition(FlowBean flowBean, Text text, int numPartitions) {
String phone=text.toString();
String prePhone=phone.substring(0,3);
if ("136".equals(prePhone)){
return 0;
}else if ("137".equals(prePhone)){
return 1;
}else if ("138".equals(prePhone)){
return 2;
}else if ("139".equals(prePhone)){
return 3;
}else {
return 4;
}
}
}
然后在driver类中再添加两行建立连接：
job.setPartitionerClass(ProvincePartitioner2.class);
job.setNumReduceTasks(5);
结果分析
可以看到有5个分区，且实现了136 137 138 139开头的手机号各在一个文件，其他手机号开头的在一个文件，而且每一个文件中实现了总流量排序，并且当总流量相同的时候再按照上行流量的正序排序。
（部分结果图片）
![[leetcode]maximum-width-of-binary-tree](https://img-blog.csdnimg.cn/img_convert/cc1543ef43f67aa5cef133b6c3f9ce13.jpeg)
