Hadoop架構之——Mapreduce WritableComparable 排序

文章目錄

- 概述
- 分類
- 自定義排序WritableComparable
- WritableComparable排序案例實操（區内排序）

概述

排序是MapReduce架構中最重要的操作之一。

MapTask和ReduceTask均會對資料按照key進行排序。該操作屬于Hadoop的預設行為。任何應用程式中的資料均會被排序，而不管邏輯上是.否需要。

預設排序是按照字典順序排序，且實作該排序的方法是快速排序。

對于MapTask,它會将處理的結果暫時放到環形緩沖區中，當環形緩沖區使用率達到一定門檻值後，再對緩沖區中的資料進行一-次快速排序,并将這些有序資料溢寫到磁盤上,而當資料處理完畢後，它會對磁盤上所有檔案進行歸并排序。

對于ReduceTask,它從每個MapTask上遠端拷貝相應的資料檔案，如果檔案大小超過一定門檻值，則溢寫磁盤上，否則存儲在記憶體中。如果磁盤上檔案數目達到-定門檻值，則進行- -次歸并排序以生成-一個更大檔案;如果記憶體中檔案大小或者數目超過一定門檻值，則進行一-次合并後将資料溢寫到磁盤上。當所有資料拷貝完畢後，ReduceTask統一對記憶體和磁盤上的所有資料進行一-次歸并排序。

分類

(1)部分排序

MapR educe根據輸入記錄的鍵對資料集排序。保證輸出的每個檔案内部有序。

(2)全排序

最終輸出結果隻有一個檔案，且檔案内部有序。實作方式是隻設定一個Reduce Task。但該方法在處理大型檔案時效率極低，因為一台機器處理所有檔案，完全喪失了MapR educe所提供的并行架構。

(3)輔助排序: (GroupingComparatar分組)

在R educe端對key進行分組。應用于:在接收的key為bean對象時，想讓一個或幾個字段相同(全部字段比較不相同)的keyi進入到同一個reduce方法時，可以采用分組排序。

(4)二次排序

在自定義排序過程中，如果compareTo中的判斷條件為兩個即為二次排序。

自定義排序WritableComparable

（1）原理分析

bean對象做為key傳輸，需要實作WritableComparable接口重寫compareTo方法，就可以實作排序

@Override
public int compareTo(FlowBean o) {

	int result;
		
	// 按照總流量大小，倒序排列
	if (sumFlow > bean.getSumFlow()) {
		result = -1;
	}else if (sumFlow < bean.getSumFlow()) {
		result = 1;
	}else {
		result = 0;
	}
	return result;
}

WritableComparable排序案例實操（全排序）

1．需求

對每個手機号的總流量進行排序。

資料如下：

13470253144	180	180	360
13509468723	7335	110349	117684
13560439638	918	4938	5856
13568436656	3597	25635	29232
13590439668	1116	954	2070
13630577991	6960	690	7650
13682846555	1938	2910	4848
13729199489	240	0	240
13736230513	2481	24681	27162
13768778790	120	120	240
13846544121	264	0	264
13956435636	132	1512	1644
13966251146	240	0	240
13975057813	11058	48243	59301
13992314666	3008	3720	6728
15043685818	3659	3538	7197
15910133277	3156	2936	6092
15959002129	1938	180	2118
18271575951	1527	2106	3633
18390173782	9531	2412	11943
84188413	4116	1432	5548

2. 代碼實作

（1）FlowBean

public class FlowBean implements WritableComparable<FlowBean> {

    private long upFlow;
    private long downFlow;
    private long sumFlow;

    public FlowBean() { }

    public FlowBean(long upFlow, long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = upFlow+downFlow;
    }

    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;
    }

    @Override
    public String toString() {
        return upFlow +
                "\t" + downFlow +
                "\t" + sumFlow ;
    }

    //TODO 核心比較條件
    @Override
    public int compareTo(FlowBean bean) {
        //TODO 核心比較條件判斷
        //TODO 正向排序法則
        int result ;  // 0 , 1 , -1  相等取0 正向取1 , 負向取-1
        if (sumFlow > bean.getSumFlow()){
            result = -1;
        }else if (sumFlow < bean.getSumFlow()){
            result = 1;
        }else {
            result = 0;
        }

        return result;
    }

    //TODO 序列化
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong( upFlow );
        out.writeLong( downFlow );
        out.writeLong( sumFlow );
    }

    //TODO 反序列化
    @Override
    public void readFields(DataInput in) throws IOException {
        upFlow = in.readLong();
        downFlow = in.readLong();
        sumFlow = in.readLong();
    }
}

（2）編寫Mapper類

ublic class FlowCountSortMapper extends Mapper<LongWritable, Text,FlowBean,Text> {

    FlowBean k = new FlowBean();
    Text v = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //TODO 擷取并切割
        String[] splits = value.toString().split( "\t" );


        //13975057813	11058	48243	59301
        //TODO 封裝對象
        String phoneNum = splits[0];


        long upFlow = Long.parseLong( splits[1] );
        long downFlow = Long.parseLong( splits[2] );
        long sumFlow = Long.parseLong( splits[3] );

        v.set( phoneNum );

        k.setUpFlow( upFlow );
        k.setDownFlow( downFlow );
        k.setSumFlow( sumFlow );

        context.write( k,v );

    }
}

（3）編寫Reducer類

public class FlowCountSortReducer extends Reducer<FlowBean, Text,Text,FlowBean> {

    @Override
    protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        for (Text value : values) {
            context.write (value,key);
        }
    }
}

（4）編寫Driver類

public class FlowCountSortDriver {
    public static void main(String[] args) throws Exception{

        // 1 擷取配置資訊，或者job對象執行個體
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        // 2 指定本程式的jar包所在的本地路徑
        job.setJarByClass(FlowCountSortDriver.class);

        // 3 指定本業務job要使用的mapper/Reducer業務類
        job.setMapperClass(FlowCountSortMapper.class);
        job.setReducerClass(FlowCountSortReducer.class);

        // 4 指定mapper輸出資料的kv類型
        job.setMapOutputKeyClass(FlowBean.class);
        job.setMapOutputValueClass(Text.class);

        // 5 指定最終輸出的資料的kv類型
        job.setOutputKeyClass( Text.class);
        job.setOutputValueClass(FlowBean.class);


        // 6 指定job的輸入原始檔案所在目錄
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 7 将job中配置的相關參數，以及job所用的java類所在的jar包， 送出給yarn去運作
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

WritableComparable排序案例實操（區内排序）

1．需求

要求每個省份手機号輸出的檔案中按照總流量内部排序。

2．需求分析

基于前一個需求，增加自定義分區類，分區按照省份手機号設定。

3．案例實操

（1）增加自定義分區類

public class ProvincePartitioner extends Partitioner<FlowBean,Text> {

    @Override
    public int getPartition(FlowBean key, Text values, int numPartitions) {
        int partition = 4;

        String substring = values.toString().substring( 0, 3 );

        if("136".equals( substring )){
            partition = 0;
        }else if("137".equals( substring )){
            partition = 1;
        }else if("138".equals( substring )){
            partition = 2;
        }else if("139".equals( substring )){
            partition = 3;
        }
        return partition;
    }
}

（2）在驅動類中添加分區類

// 加載自定義分區類
job.setPartitionerClass(ProvincePartitioner.class);
// 設定Reducetask個數
job.setNumReduceTasks(5);

Hadoop架構之——Mapreduce WritableComparable 排序

文章目錄

概述

分類

自定義排序WritableComparable

WritableComparable排序案例實操（區内排序）

繼續閱讀

申請評分模型拒絕推斷（RI）方法申請評分模型拒絕推斷（RI）方法

專家訪談：搜尋開源力量：Lucene技術前景

Sql優化一：sql語句優化

Nacos 2.0 更新前後性能對比壓測

hadoop 用MR實作join操作

Centos7 下 Hadoop 2.6.4 分布式叢集環境搭建摘要叢集準備安裝JDK 安裝 Hadoop 2.6.4 部署 slaver1-slaver4 啟動 hadoop 叢集成功了

尚矽谷—韓順平—圖解 Java設計模式（結構型）（55～）

Storm編譯打包過程中遇到的一些問題及解決方法

MapReduce的幾個企業級經典面試案例MapReduce的幾個企業級經典面試案例

9.spark Core 進階2--Cashe

淺談企業活動中進行資料分析的重要性

ubuntu14.04下安裝hbse1.0.1.1

User Defined Hadoop DataType

Ambari介紹和架構原理

NOSQL安全攻擊

win10本地scala和spark安裝安裝scala安裝spark