Hadoop之Shuffle機(jī)制詳解

---

目錄

Shuffle機(jī)制

Partition分區(qū)

WritableComparable排序

Combiner合并

GroupingComparator分組（輔助排序）

---

1. Shuffle機(jī)制

Mapreduce確保每個(gè)reducer的輸入都是按key排序的。系統(tǒng)執(zhí)行排序的過程（即將mapper輸出作為輸入傳給reducer）稱為shuffle，如下圖所示

---

2. Partition分區(qū)

問題引出：要求將統(tǒng)計(jì)結(jié)果按照條件輸出到不同文件中（分區(qū)）。比如：將統(tǒng)計(jì)結(jié)果按照手機(jī)歸屬地不同省份輸出到不同文件中（分區(qū)）

默認(rèn)partition分區(qū)

public class HashPartitioner<K, V> extends Partitioner<K, V> {public int getPartition(K key, V value, int numReduceTasks) {return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;} } 默認(rèn)分區(qū)是根據(jù)key的hashCode對(duì)reduceTasks個(gè)數(shù)取模得到的。用戶沒法控制哪個(gè)key存儲(chǔ)到哪個(gè)分區(qū)。

---

自定義Partitioner步驟

自定義類繼承Partitioner，重寫getPartition()方法

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {@Overridepublic int getPartition(Text key, FlowBean value, int numPartitions) {// 1 獲取電話號(hào)碼的前三位String preNum = key.toString().substring(0, 3);int partition = 4;// 2 判斷是哪個(gè)省if ("136".equals(preNum)) {partition = 0;}else if ("137".equals(preNum)) {partition = 1;}else if ("138".equals(preNum)) {partition = 2;}else if ("139".equals(preNum)) {partition = 3;}return partition;} }

在job驅(qū)動(dòng)中，設(shè)置自定義partitioner：

job.setPartitionerClass(CustomPartitioner.class);

自定義partition后，要根據(jù)自定義partitioner的邏輯設(shè)置相應(yīng)數(shù)量的reduce task

job.setNumReduceTasks(5);

---

注意

如果reduceTask的數(shù)量> getPartition的結(jié)果數(shù)，則會(huì)多產(chǎn)生幾個(gè)空的輸出文件part-r-000xx；

如果1<reduceTask的數(shù)量<getPartition的結(jié)果數(shù)，則有一部分分區(qū)數(shù)據(jù)無處安放，會(huì)Exception；

如果reduceTask的數(shù)量=1，則不管mapTask端輸出多少個(gè)分區(qū)文件，最終結(jié)果都交給這一個(gè)reduceTask，最終也就只會(huì)產(chǎn)生一個(gè)結(jié)果文件 part-r-00000；

例如：假設(shè)自定義分區(qū)數(shù)為5，則
（1）job.setNumReduceTasks(1);會(huì)正常運(yùn)行，只不過會(huì)產(chǎn)生一個(gè)輸出文件
（2）job.setNumReduceTasks(2);會(huì)報(bào)錯(cuò)
（3）job.setNumReduceTasks(6);大于5，程序會(huì)正常運(yùn)行，會(huì)產(chǎn)生空文件

---

3. WritableComparable排序

排序是MapReduce框架中最重要的操作之一。Map Task和Reduce Task均會(huì)對(duì)數(shù)據(jù)（按照key）進(jìn)行排序。該操作屬于Hadoop的默認(rèn)行為。任何應(yīng)用程序中的數(shù)據(jù)均會(huì)被排序，而不管邏輯上是否需要。默認(rèn)排序是按照字典順序排序，且實(shí)現(xiàn)該排序的方法是快速排序。

對(duì)于Map Task，它會(huì)將處理的結(jié)果暫時(shí)放到一個(gè)緩沖區(qū)中，當(dāng)緩沖區(qū)使用率達(dá)到一定閾值后，再對(duì)緩沖區(qū)中的數(shù)據(jù)進(jìn)行一次排序，并將這些有序數(shù)據(jù)寫到磁盤上，而當(dāng)數(shù)據(jù)處理完畢后，它會(huì)對(duì)磁盤上所有文件進(jìn)行一次合并，以將這些文件合并成一個(gè)大的有序文件。

對(duì)于Reduce Task，它從每個(gè)Map Task上遠(yuǎn)程拷貝相應(yīng)的數(shù)據(jù)文件，如果文件大小超過一定閾值，則放到磁盤上，否則放到內(nèi)存中。如果磁盤上文件數(shù)目達(dá)到一定閾值，則進(jìn)行一次合并以生成一個(gè)更大文件；如果內(nèi)存中文件大小或者數(shù)目超過一定閾值，則進(jìn)行一次合并后將數(shù)據(jù)寫到磁盤上。當(dāng)所有數(shù)據(jù)拷貝完畢后，Reduce Task統(tǒng)一對(duì)內(nèi)存和磁盤上的所有數(shù)據(jù)進(jìn)行一次合并。

每個(gè)階段的默認(rèn)排序

排序的分類

部分排序：區(qū)內(nèi)排序——環(huán)形緩沖區(qū)
MapReduce根據(jù)輸入記錄的鍵對(duì)數(shù)據(jù)集排序。保證輸出的每個(gè)文件內(nèi)部排序。

全排序：
如何用Hadoop產(chǎn)生一個(gè)全局排序的文件？最簡(jiǎn)單的方法是使用一個(gè)分區(qū)。但該方法在處理大型文件時(shí)效率極低，因?yàn)橐慌_(tái)機(jī)器必須處理所有輸出文件，從而完全喪失了MapReduce所提供的并行架構(gòu)。
替代方案：首先創(chuàng)建一系列排好序的文件；其次，串聯(lián)這些文件；最后，生成一個(gè)全局排序的文件。主要思路是使用一個(gè)分區(qū)來描述輸出的全局排序。例如：可以為上述文件創(chuàng)建3個(gè)分區(qū)，在第一分區(qū)中，記錄的單詞首字母a-g，第二分區(qū)記錄單詞首字母h-n, 第三分區(qū)記錄單詞首字母o-z。

輔助排序：（GroupingComparator分組）
Mapreduce框架在記錄到達(dá)reducer之前按鍵對(duì)記錄排序，但鍵所對(duì)應(yīng)的值并沒有被排序。甚至在不同的執(zhí)行輪次中，這些值的排序也不固定，因?yàn)樗鼈儊碜圆煌膍ap任務(wù)且這些map任務(wù)在不同輪次中完成時(shí)間各不相同。一般來說，大多數(shù)MapReduce程序會(huì)避免讓reduce函數(shù)依賴于值的排序。但是，有時(shí)也需要通過特定的方法對(duì)鍵進(jìn)行排序和分組等以實(shí)現(xiàn)對(duì)值的排序。

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

自定義排序WritableComparable
（1）原理分析
bean對(duì)象實(shí)現(xiàn)WritableComparable接口重寫compareTo方法，就可以實(shí)現(xiàn)排序

@Override public int compareTo(FlowBean o) {// 倒序排列，從大到小return this.sumFlow > o.getSumFlow() ? -1 : 1; }

---

4. Combiner合并

combiner是MR程序中Mapper和Reducer之外的一種組件。

combiner組件的父類就是Reducer。

combiner和reducer的區(qū)別在于運(yùn)行的位置：

Combiner是在每一個(gè)maptask所在的節(jié)點(diǎn)運(yùn)行;

Reducer是接收全局所有Mapper的輸出結(jié)果；

combiner的意義就是對(duì)每一個(gè)maptask的輸出進(jìn)行局部匯總，以減小網(wǎng)絡(luò)傳輸量。

combiner能夠應(yīng)用的前提是不能影響最終的業(yè)務(wù)邏輯，而且，combiner的輸出kv應(yīng)該跟reducer的輸入kv類型要對(duì)應(yīng)起來。

自定義Combiner實(shí)現(xiàn)步驟：

自定義一個(gè)combiner繼承Reducer，重寫reduce方法

public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{@Overrideprotected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {// 1 匯總操作int count = 0;for(IntWritable v :values){count += v.get();}// 2 寫出context.write(key, new IntWritable(count));} }

在job驅(qū)動(dòng)類中設(shè)置：

job.setCombinerClass(WordcountCombiner.class);

---

5. GroupingComparator分組（輔助排序）

對(duì)reduce階段的數(shù)據(jù)根據(jù)某一個(gè)或幾個(gè)字段進(jìn)行分組。

總結(jié)

以上是生活随笔為你收集整理的Hadoop之Shuffle机制详解的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。