hadoop 實作資料排序-有解無憂

前言

在很多業務場景下，需要對原始的資料讀取分析后，將輸出的結果按照指定的業務欄位進行排序輸出，方便上層應用對結果資料進行展示或使用，減少二次排序的成本

在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個磁區檔案下，并且每個磁區檔案內的總流量也是按照從高到低的順序
在這里插入圖片描述

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/403913.html

標籤：其他

上一篇：hadoop完全分布式搭建（最通俗易懂）

下一篇：zookeeper 安裝部署與簡單使用