一、mr的shuffle流程

說到mr的Shuffle，那麼首先要明確shuffle屬於哪個階段。shuffle就是從Map端輸出到Reduce輸入的整個過程，這個過程廣義上稱之為shuffle。Shuffle橫跨Map端和Reduce端，在Map端包括Spill過程，在Reduce端包括copy和sort過程，如圖所示：

Map的shuffle過程包括了spill、collect、sort、和Marge

首先每個map任務不斷地以鍵值對的形式，把資料輸出到記憶體中構造的一個環形資料結構中，這樣可以有效地使用記憶體空間，在記憶體中放置儘可能過得資料，之後把資料安排partition值和key進行升序排序，這裡只是移動了索引資料，其次是spill執行緒，會為spill過程建立一個磁碟檔案，只有會將partition的資料放到這個檔案中，一直把所有的partition遍歷完。如果我們map任務輸出資料量太大的話，可能會進行好幾次的spill，這個時候會把這些檔案進行合併，也就是merge過程。

Reduce的shuffle過程包括了copy、merge sort

reduce任務通過http向各個map任務拖取所需要的資料，其中一項就是響應reduce拖取map資料，當有mapoutput的請求過來時，就會讀取響應的map輸出檔案中對應的這個reduce部分的資料，輸出給到reduce，之後的megre和map端的megre過程一樣，因為map的輸出資料已經是有序的，megre進行一次合併排序，當reduce輸入檔案已經定了，那麼整個shuffle才最終結束

二、hivesql優化的點

** 優化的根本思想：**

• 儘早儘量過濾資料，減少每個階段的資料量

• 減少job數

• 解決資料傾斜問題

1. 列剪裁

   儘量不適用select * from 表名，而是根據我們所需要的欄位去展示，例如我們的test表中有 a，b，c三個欄位，我們只需要a與c的時候我們使用

   select a,b from test                -----建議使用
   select * from test                 -----不建議使用

2. 分割槽剪裁

   在查詢的過程中減少不必要的分割槽，即儘量指定分割槽

3. 利用hive的優化機制減少job數

   不論是什麼關聯，如果join的key相同，不管有多少表，都會合併為一個MapReducer任 務，所以在多表join時儘量用到相同的欄位

4. 善用union all

   對於同一張表的union all要比多重的insert快的多，因為hive本身對這種union all做過優化，即可以只掃描一次源表

5. join前過濾掉不需要的資料

   在我們進行join操作的時候可能會出現要join的欄位為空或者說是無效，這樣會產生傾斜問題，我們可以把空值的key變成一個字串加上隨機數，這樣能把傾斜的資料分到不同的reduce上，解決資料傾斜問題

   select *
       from A a
   left outer join B b
       on case when a.id is null then concat('隨意字串',rand()) else a.id=b.id

6. 小表放前大表放後

   在編寫帶有join的程式碼語句時，應該將條目少的表/子查詢放在join操作符的前面

   因為在Reduce階段，位於join操作符左邊的表會先被載入到記憶體，載入條目較少的表可以有效的防止記憶體溢位(OOM)。所以對於同一個key來說，對應的value值小的放前面，大的放後面

7. 使用動態分割槽

8. 更換hive底層引擎

   hive的底層引擎是mr，我們可以將底層引擎換成tez或者是spark等，提升效率

三、資料傾斜解決方法

資料傾斜對於分散式大資料系統來將是非常可怕的。

在理想的狀態下，隨著節點數量的增加，應用整體耗時線性下降。如果一臺機器處理大量資料需要60分鐘，那麼當機器數量增加到三臺時，最理想的耗時是20分鐘。如果想要做到這些，就必須要保證每臺機器的任務量相等，但很多時候任務的分配是不均勻的，這個就是分散式環境下最大的問題。那我們如何緩解解決資料傾斜呢？

1、過濾異常的資料

如果導致資料清洗的key是異常資料，那麼簡單的過濾掉就可以了。

首先要對key進行分析，判斷是哪些key導致的資料傾斜，我們可以通過抽樣統計kry的出現次數來驗證。

def.select("key").smaple(false,0.1)                 //資料取樣
    .(k => (k,1)).reduceByKey(_ + _)                //統計key出現的次數
    .map(k => (k._2,k._1)).sortByKey(false)         //根據key出現的次數進行排序
    .take(N)                                        //提取前N個

解決的方案則是對資料進行過濾即可

2、提高shuffle並行度

調整shuffle時的並行度，使得原本被分配到同一個Task的不同Key發配到不同的Task上處理，降低原Task所需處理的資料量，從而緩解資料傾斜造成的短板問題

3、自定義 Partitioner

使用自定義的 Partitioner（預設為 HashPartitioner），將原本被分配到同一個 Task 的不同 Key 分配到不同 Task。

4、大表 key 加鹽，小表擴大 N 倍 jion

如果出現數據傾斜的 Key 比較多，上一種方法將這些大量的傾斜 Key 分拆出來，意義不大。此時更適合直接對存在資料傾斜的資料集全部加上隨機字首，然後對另外一個不存在嚴重資料傾斜的資料集整體與隨機字首集作笛卡爾乘積（即將資料量擴大N倍）。