亲宝软件园·资讯

展开

hadoop 数据排序

小码农叔叔 人气:0

前言

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

MapReduce排序

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点:

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

加载全部内容

相关教程
猜你喜欢
用户评论