MapReduce 資料傾斜原因和解決方案
MapReduce是面向大資料並行處理的計算模型、框架和平臺，它隱含了以下三層含義：
1）MapReduce是一個基於叢集的高效能平行計算平臺（Cluster Infrastructure）。它允許用市場上普通的商用伺服器構成一個包含數十、數百至數千個節點的分佈和平行計算叢集。
2）MapReduce是一個平行計算與執行軟體框架（Software Framework）。它提供了一個龐大但設計精良的平行計算軟體框架，能自動完成計算任務的並行化處理，自動劃分計算資料和計算任務，在叢集節點上自動分配和執行任務以及收集計算結果，將資料分佈儲存、資料通訊、容錯處理等平行計算涉及到的很多系統底層的複雜細節交由系統負責處理，大大減少了軟體開發人員的負擔。
3）MapReduce是一個並行程式設計模型與方法（Programming Model & Methodology）。它藉助於函式式程式設計語言Lisp的設計思想，提供了一種簡便的並行程式設計方法，用Map和Reduce兩個函式程式設計實現基本的平行計算任務，提供了抽象的操作和並行程式設計介面，以簡單方便地完成大規模資料的程式設計和計算處理。
什麼是資料傾斜及資料傾斜是怎麼產生
簡單來說資料傾斜就是資料的key 的分化嚴重不均，造成一部分資料很多，一部分資料很少的局面。
舉個 word count 的入門例子，它的map 階段就是形成 （“aaa”,1）的形式，然後在reduce 階段進行 value 相加，得出 “aaa” 出現的次數。若進行 word count 的文字有100G，其中 80G 全部是 “aaa” 剩下 20G 是其餘單詞，那就會形成 80G 的資料量交給一個 reduce 進行相加，其餘 20G 根據 key 不同分散到不同 reduce 進行相加的情況。如此就造成了資料傾斜，臨床反應就是 reduce 跑到 99%然後一直在原地等著 那80G 的reduce 跑完。
如下圖：
這樣就能清楚看到，資料經過 map後，由於不同key 的資料量分佈不均，在shuffle 階段中通過 partition 將相同的 key 的資料打上發往同一個 reducer 的標記，然後開始 spill （溢寫）寫入磁碟，最後merge成最終map階段輸出檔案。需要大資料資料的可以加群：834325294
如此一來 80G 的 aaa 將發往同一個 reducer ，由此就可以知道 reduce 最後 1% 的工作在等什麼了。
為什麼說資料傾斜與業務邏輯和資料量有關
從另外角度看資料傾斜，其本質還是在單臺節點在執行那一部分資料reduce任務的時候，由於資料量大，跑不動，造成任務卡住。若是這臺節點機器記憶體夠大，CPU、網路等資源充足，跑 80G 左右的資料量和跑10M 資料量所耗時間不是很大差距，那麼也就不存在問題，傾斜就傾斜吧，反正機器跑的動。所以機器配置和資料量存在一個合理的比例，一旦資料量遠超機器的極限，那麼不管每個key的資料如何分佈，總會有一個key的資料量超出機器的能力，造成 reduce 緩慢甚至卡頓。
業務邏輯造成的資料傾斜會多很多，日常使用過程中，容易造成資料傾斜的原因可以歸納為幾點：
容易造成資料傾斜的原因
分組 注：group by 優於distinct group
情形：group by 維度過小，某值的數量過多
後果：處理某值的reduce非常耗時
去重 distinct count(distinct xx)
情形：某特殊值過多
後果：處理此特殊值的reduce耗時
連線 join
情形1：其中一個表較小，但是key集中
後果1：分發到某一個或幾個Reduce上的資料遠高於平均值
情形2：大表與大表，但是分桶的判斷欄位0值或空值過多
後果2：這些空值都由一個reduce處理，非常慢
資料傾斜的影響
hadoop 中資料傾斜會極大影響效能和效率。
資料分佈（導致資料傾斜）
正常的資料分佈理論上都是傾斜的，就是我們所說的20-80原理：80%的財富集中在20%的人手中, 80%的使用者只使用20%的功能 , 20%的使用者貢獻了80%的訪問量 , 不同的資料欄位可能的資料傾斜一般有兩種情況:
一種是唯一值非常少，極少數值有非常多的記錄值(唯一值少於幾千)
一種是唯一值比較多，這個欄位的某些值有遠遠多於其他值的記錄數，但是它的佔比也小於百分之一或千分之一
或是這麼說：
1. 資料頻率傾斜——某一個區域的資料量要遠遠大於其他區域。
2. 資料大小傾斜——部分記錄的大小遠遠大於平均值。
分割槽
常見的mapreduce分割槽方式為hash 和range ，
hash partition 的好處是比較彈性，跟資料型別無關，實現簡單（設定reduce個數就好，一般不需要自己實現）
range partition 需要實現者自己瞭解資料分佈, 有時候需要手工做sample取樣. 同時也不夠彈性, 表現在幾個方面：
1. 對同一個表的不同欄位都需要實現不同的range partition, 對於時間這種欄位根據查詢型別的不同或者過濾條件的不同切分range 的大小都不一定.
2 .有時候可能設計使用多個欄位組合的情況, 這時候又不能使用之前單個欄位的partition 類, 並且多個欄位組合之間有可能有隱含的聯絡，比如出生日期和星座，商品和季節.
3. 手工做sample 非常耗時間，需要使用者對查詢使用的資料集的分佈有領域知識.
4. 分配方式是死的，reduce 個數是確定的，一旦某種情況下發生傾斜，調整引數
其他的分割槽型別還有hbase 的hregionpartitioner 或者totalorder partitioner 等.
解決方案一
1、調優引數
set hive.map.aggr=true；
set hive.groupby.skewindata=true;
hive.map.aggr=true：在map中會做部分聚集操作，效率更高但需要更多的記憶體。
hive.groupby.skewindata=true：資料傾斜時負載均衡，當選項設定為true，生成的查詢計劃會有兩個MRJob。第一個MRJob 中，Map的輸出結果集合會隨機分佈到Reduce中，每個Reduce做部分聚合操作，並輸出結果，這樣處理的結果是相同的GroupBy Key有可能被分發到不同的Reduce中，從而達到負載均衡的目的；第二個MRJob再根據預處理的資料結果按照GroupBy Key分佈到Reduce中（這個過程可以保證相同的GroupBy Key被分佈到同一個Reduce中），最後完成最終的聚合操作。
由上面可以看出起到至關重要的作用的其實是第二個引數的設定，它使計算變成了兩個mapreduce，先在第一個中在 shuffle 過程 partition 時隨機給 key 打標記，使每個key 隨機均勻分佈到各個 reduce 上計算，但是這樣只能完成部分計算，因為相同key沒有分配到相同reduce上，所以需要第二次的mapreduce,這次就回歸正常 shuffle,但是資料分佈不均勻的問題在第一次mapreduce已經有了很大的改善，因此基本解決資料傾斜。
2、在 key 上面做文章，在 map 階段將造成傾斜的key 先分成多組，例如 aaa 這個 key,map 時隨機在 aaa 後面加上 1,2,3,4 這四個數字之一，把 key 先分成四組，先進行一次運算，之後再恢復 key 進行最終運算。
3、能先進行 group 操作的時候先進行 group 操作，把 key 先進行一次 reduce,之後再進行 count 或者 distinct count 操作。
4、join 操作中，使用 map join 在 map 端就先進行 join ，免得到reduce 時卡住。
以上4中方式，都是根據資料傾斜形成的原因進行的一些變化。要麼將 reduce 端的隱患在 map 端就解決，要麼就是對 key 的操作，以減緩reduce 的壓力。總之瞭解了原因再去尋找解決之道就相對思路多了些，方法肯定不止這4種。
看了其他的部落格
解決方案二
1. 增加reduce 的jvm記憶體
2. 增加reduce 個數
3. customer partition
4. 其他優化的討論.
5. reduce sort merge排序演算法的討論
6. 正在實現中的hive skewed join.
7. pipeline
8. distinct
9. index 尤其是bitmap index
方式1
既然reduce 本身的計算需要以合適的記憶體作為支援，在硬體環境容許的情況下，增加reduce 的記憶體大小顯然有改善資料傾斜的可能，這種方式尤其適合資料分佈第一種情況，單個值有大量記錄, 這種值的所有紀錄已經超過了分配給reduce 的記憶體，無論你怎麼樣分割槽這種情況都不會改變. 當然這種情況的限制也非常明顯,
1. 記憶體的限制存在
2. 可能會對叢集其他任務的執行產生不穩定的影響.
方式2
這個對於資料分佈第二種情況有效，唯一值較多，單個唯一值的記錄數不會超過分配給reduce 的記憶體. 如果發生了偶爾的資料傾斜情況，增加reduce 個數可以緩解偶然情況下的某些reduce 不小心分配了多個較多記錄數的情況. 但是對於第一種資料分佈無效.
方式3
一種情況是某個領域知識告訴你資料分佈的顯著型別，比如hadoop definitive guide 裡面的溫度問題，一個固定的組合(觀測站點的位置和溫度) 的分佈是固定的, 對於特定的查詢如果前面兩種方式都沒用，實現自己的partitioner 也許是一個好的方式.
方式5
reduce 分配的記憶體遠小於處理的資料量時，會產生multi-pass sort 的情況是瓶頸，那麼就要問
1. 這種排序是有必要的嘛？
2. 是否有其他排序演算法或優化可以根據特定情況降低他瓶頸的閾值?
3. map reduce 適合處理這種情況嘛?
關於問題1. 如果是group by , 那麼對於資料分佈情況1 ,hash 比sort 好非常多，即使某一個reduce 比其他reduce 處理多的多的資料，hash 的計算方式也不會差距太大.
問題2. 一個是如果實現block shuffle 肯定會極大的減少排序本身的成本, 另外，如果分割槽之後的reduce 不是使用copy –> sort-merge –> reduce 的計算方式, 在copy 之後將每個block 的頭部資訊儲存在記憶體中，不用sort – merge 也可以直接計算reduce, 只不過這時候變成了隨機訪問，而不是現在的sort-merge 之後的順序訪問. block shuffle 的實現有兩種型別，一種是當hadoop 中真正有了列資料格式的時候，資料有更大的機會已經排過序並且按照block 來切分,一般block 為1M ( 可以關注avro-806 ) , 這時候的mapper 什麼都不做，甚至連計算分割槽的開銷都小了很多倍，直接進入reduce 最後一步，第二種型別為沒有列資料格式的支援，需要mapper 排序得到之後的block 的最大最小值，reduce 端在記憶體中儲存最大最小值，copy 完成後直接用這個值來做隨機讀然後進行reduce. ( block shuffle 的實現可以關注 MAPREDUCE-4039 ， hash 計算可以關注 MAPREDUCE-1639)
問題3 . map reduce 只有兩個函式，一個map 一個 reduce, 一旦發生資料傾斜就是partition 失效了，對於join 的例子，某一個key 分配了過多的記錄數，對於只有一次partittion的機會，分配錯了資料傾斜的傷害就已經造成了，這種情況很難除錯，但是如果你是基於map-reduce-reduce 的方式計算，那麼對於同一個key 不需要分配到同一個reduce 中，在第一個reduce 中得到的結果可以在第二個reduce 才彙總去重，第二個reduce 不需要sort – merge 的步驟，因為前一個reduce 已經排過序了，中間的reduce 處理的資料不用關心partition 怎麼分，處理的資料量都是一樣大，而第二個reduce 又不使用sort-merge 來排序，不會遇到現在的記憶體大小的問題，對於skewed join 這種情況瓶頸自然小很多.
方式6
目前hive 有幾個正在開發中的處理skewed join 情況的jira case, HIVE-3086 , HIVE-3286 ,HIVE-3026 . 簡單介紹一下就是facebook 希望通過手工處理提前列舉的方式列出單個傾斜的值，在join 的時候將這些值特殊列出當作map join 來處理，對於其他值使用原來的方式. 我個人覺得這太不伸縮了，值本身沒有考慮應用過濾條件和優化方式之後的資料量大小問題，他們提前列出的值都是基於整個分割槽的. join key 如果為組合key 的情況也應該沒有考慮，對metastore 的儲存問題有限制，對輸入的大表和小表都會scan 兩次( 一次處理非skew key , 一次處理skew key 做map join), 對輸出表也會scan 兩次（將兩個結果進行merge） , skew key 必須提前手工列出這又存在額外維護的成本，目前因為還沒有完整的開發完到能夠投入生產的情況，所以等所有特性處理完了有了文件在看看這個處理方式是否有效，我個人認為的思路應該是接著bucked map join 的思路往下走，只不過不用提前處理cluster key 的問題, 這時候cluster key 的選擇應該是join key + 某個能分散join key 的列, 這等於將大表的同一個key 的值分散到了多個不同的reduce 中，而小表的join key 也必須cluster 到跟大表對應的同一個key , join 中對於資料分佈第二種情況不用太難，增加reduce 個數就好，主要是第一種，需要大表的join key 能夠分散，對於同樣join key 的小表又能夠匹配到所有大表中的記錄. 這種思路就是不用掃描大表兩遍或者結果輸出表，不需要提前手工處理，資料是動態sample 的應用了過濾條件之後的資料，而不是提前基於統計資料的不準確結果. 這個基本思路跟tenzing 裡面描述的distributed hash join 是一樣的，想辦法切成合適的大小然後用hash 和 map join .
方式7
當同時出現join 和group 的時候, 那麼這兩個操作應該是以pipeline (管道) 的方式執行. 在join 的時候就可以直接使用group 的操作符減少大量的資料，而不是等待join 完成，然後寫入磁碟，group 又讀取磁碟做group操作. HIVE-2206 正在做這個優化. hive 裡面是沒有pipeline 這個概念的. 像是cloudera 的crunch 或者twitter 的Scalding 都是有這種概念的.
方式8
distinct 本身就是group by 的一種簡寫，我原先以為count(distinct x)這種跟group by 是一樣的，但是發現hive 裡面distinct 明顯比group by 要慢，可能跟group by 會有map 端的combiner有關, 另外觀察到hive 在預估count(distinct x) 的reduce 個數比group by 的個數要少 , 所以hive 中使用count(distinct x) , 要麼儘量把reduce 個數設定大，直接設定reduce 個數或者hive.exec.reducers.bytes.per.reducer 調小，我個人比較喜歡調後面一個,hive 目前的reduce 個數沒有統計資訊的情況下就是用map端輸入之前的數值, 如果你是join 之後還用count(distinct x) 的話，這個預設值一般都會悲劇，如果有where 條件並能過濾一定數量的資料，那麼預設reduce 個數可能就還好一點. 不管怎樣，多浪費一點reduce slot 總比等十幾甚至幾十分鐘要好, 或者轉換成group by 的寫法也不錯，寫成group by 的時候distributed by 也很有幫助.
方式9
hive 中的index 就是物化檢視，對於group by 和distinct 的情況等於變成了map 端在做計算，自然不存在傾斜. 尤其是bitmap index , 對於唯一值比較少的列優勢更大，不過index 麻煩的地方在於需要判斷你的sql 是不是常用sql , 另外如果create index 的時候沒有選你查詢的時候用的欄位，這個index 是不能用的( hive 中是永遠不可能有DBMS中的用index 去lookup 或者join 原始表這種概念的)
總結
資料傾斜沒有一勞永逸的方式可以解決，瞭解你的資料集的分佈情況，然後瞭解你所使用計算框架的執行機制和瓶頸，針對特定的情況做特定的優化，做多種嘗試，觀察是否有效.
大家喜歡多多關注，會不定期更新，你們的關注是我最大的動力。