hadoop 详解如何实现数据排序

hadoop 详解如何实现数据排序

目录前言MapReduce排序MapReduce排序分类1、部分排序2、全排序3、辅助排序4、二次排序自定义排序案例1、自定义一个Bean对象，实现WritableComparable接口2、自定义Mapper3、自定义Reducer4、自定义Driver类分区内排序案例1、添加自定义分区2、改造Driver类

前言

在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 {

//峰值流量

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 0http://;

}

}

//提供无参构造

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 {

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 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 {

@Override

protected void reduce(PhoneSortBean key, Iterable 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 设置Mahttp://p端输出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.scFiXRFVruumFlow > 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 {

@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个分区文件下，并且每个分区文件内的总流量也是按照从高到低的顺序

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。