數據傾斜的原因

1）key分布不均勻，本質上就是業務數據有可能會存在傾斜

2）某些SQL語句本身就有數據傾斜?????????????

關鍵詞	情形	后果
Join	A、其中一個表較小，但是key集中; B、兩張表都是大表，key不均	分發到某一個或幾個Reduce上的數據遠高于平均值
group by	group by 維度過小，某值的數量過多	處理某值的reduce非常耗時
Count(distinct )	某值的數量過多，多次distinct	處理某值的reduce非常耗時，造成數據膨脹

數據產生傾斜的原理：

數據傾斜問題，通常是指參與計算的數據分布不均，即某個key或者某些key的數據量遠超其他key，導致在shuffle階段，大量相同key的數據被發往同一個Reduce，進而導致該Reduce所需的時間遠超其他Reduce，成為整個任務的瓶頸。

數據傾斜的表現

?任務進度長時間維持在99%（或100%），查看任務監控頁面，發現只有少量（1個或幾個）reduce子任務未完成。因為其處理的數據量和其他reduce差異過大。

?單一reduce的記錄數與平均記錄數差異過大，通常可能達到3倍甚至更多。 最長時長遠大于平均時長。

處理數據傾斜

?group by 產生傾斜的問題

前面文章已經提過開啟map端聚合，對于開啟map端聚合后，數據會現在Map端完成部分聚合工作。這樣一來即便原始數據是傾斜的，經過Map端的初步聚合后，發往Reduce的數據也就不再傾斜了。最佳狀態下，Map-端聚合能完全屏蔽數據傾斜問題。group by 產生傾斜的問題。

但是并不是所有map端聚合都能完全屏蔽，我們用到結合另外另外一個參數一起使用，開啟負載均衡。

開啟map聚合：

set hive.map.aggr=true; （默認是?true）

開啟map端combiner，減少reduce 拉取的數據量。

set hive.groupby.skewindata=true; （默認是?false）

原理是生成的查詢計劃會有兩個 MR Job。第一個 MR Job 中，Map 的輸出結果集合會隨機分布到 Reduce 中，每個 Reduce 做部分聚合操作，并輸出結果，這樣處理的結果是相同的 Group By Key 有可能被分發到不同的 Reduce 中，從而達到負載均衡的目的；第二個 MR Job 再根據預處理的數據結果按照 Group By Key 分布到 Reduce 中（這個過程可以保證相同的 Group By Key 被分布到同一個 Reduce 中），最后完成最終的聚合操作。

實例

explain select province_id??????? ????????,count(*) as cnt ???from ds_hive.ch12_order_detail_orc t1 ???group by province_id ;

第一種情況（關閉map聚合和負載均衡），會將3857w數據全部傳到reduce階段進行group by聚合，執行時間60.8s

后臺日志說明，map-reduce數據轉移處理了8000w條，效率低

第二種（開啟map聚合和關閉負載均衡）執行時間26s

后臺日志說明，map-reduce數據轉移處理了340條，效率高

第三種（開啟map聚合和負載均衡）執行時間45s

job1：Map 的輸出結果集合會隨機分布到 Reduce 中，每個 Reduce 做部分聚合操作，并輸出結果

job2:將job1預處理后的數據再進行group by，完成聚合

后臺日志說明，job1的map-reduce數據轉移處理了340條，但沒有完成最終分組聚合，而是將多個reduce數據傳給job2的map-reduce進行處理，一定程度上增加了資源的消耗和處理，所以效率比map聚合的一次mr效率低

Join導致的數據傾斜

前文提到過，未經優化的join操作，默認是使用common join算法，也就是通過一個MapReduce Job完成計算。Map端負責讀取join操作所需表的數據，并按照關聯字段進行分區，通過Shuffle，將其發送到Reduce端，相同key的數據在Reduce端完成最終的Join操作。

如果關聯字段的值分布不均，就可能導致大量相同的key進入同一Reduce，從而導致數據傾斜問題。

由join導致的數據傾斜問題，有如下三種解決方案：

map join

使用map join算法，join操作僅在map端就能完成，沒有shuffle操作，沒有reduce階段，自然不會產生reduce端的數據傾斜。該方案適用于大表join小表時發生數據傾斜的場景。

相關參數如下：

--啟動Map Join自動轉換 set hive.auto.convert.join=true; --一個Common Join operator轉為Map Join operator的判斷條件,若該Common Join相關的表中,存在n-1張表的大小總和<=該值,則生成一個Map Join計劃,此時可能存在多種n-1張表的組合均滿足該條件,則hive會為每種滿足條件的組合均生成一個Map Join計劃,同時還會保留原有的Common Join計劃作為后備(back up)計劃,實際運行時,優先執行Map Join計劃，若不能執行成功，則啟動Common Join后備計劃。 set hive.mapjoin.smalltable.filesize=250000; --開啟無條件轉Map Join set hive.auto.convert.join.noconditionaltask=true; --無條件轉Map Join時的小表之和閾值,若一個Common Join operator相關的表中，存在n-1張表的大小總和<=該值,此時hive便不會再為每種n-1張表的組合均生成Map Join計劃,同時也不會保留Common Join作為后備計劃。而是只生成一個最優的Map Join計劃。 set hive.auto.convert.join.noconditionaltask.size=10000000; hive (default)> create table ds_hive.ch12_order_detail_qingxie_join1 as ????SELECT t1.id ???????????,t2.province_name ??????from ds_hive.ch12_order_detail_orc t1 ?left join ds_hive.ch12_province_info_orc? t2 ????????on t1.province_id=t2.id ;

province_id字段是存在傾斜的，若不經過優化，執行日志顯示一直卡在99-100直接，單一任務執行了2m35s，這倆種表現都能說明出現了數據傾斜問題，最終執行時間188s

開啟mapjoin的自動轉換后，可以看出只有map階段而沒有reduce階段，最終執行時間為26s，相較188s，速度大大增快

skew join

并不是所有的join都能滿足mapjoin,這里我們又出現一種優化方案，那就是skew,skew join的原理是，為傾斜的大key單獨啟動一個map join任務進行計算，其余key進行正常的common join。但是他只能只能某種特定的innerjoin優化，原理圖如下：

相關參數如下：

--啟用skew join優化 set hive.optimize.skewjoin=true; --觸發skew join的閾值，若某個key的行數超過該參數值，則觸發 set hive.skewjoin.key=1000000; 這種方案對參與join的源表大小沒有要求，但是對兩表中傾斜的key的數據量有要求，要求一張表中的傾斜key的數據量比較小（方便走mapjoin）。 set hive.auto.convert.join=true; set hive.optimize.skewjoin=true; explain create table ds_hive.ch12_order_detail_qingxie_join1 as ????SELECT t1.id ???????????,t2.province_name ??????from ds_hive.ch12_order_detail_orc t1 ???????join ds_hive.ch12_province_info_orc? t2 ????????on t1.province_id=t2.id ;

調整SQL語句

若參與join的兩表均為大表，其中一張表的數據是傾斜的，此時也可通過以下方式對SQL語句進行相應的調整。

1）如果是空值關聯造成的傾斜，把空值的key變成一隨機數(隨機值類型需要跟key的類型一致)，把傾斜的數據分到不同的reduce上，由于null值關聯不上，處理后并不影響最終結果。注意：join的字段類型一定要一致，否則數據不會分到不同的reduce上。

2）如果熱點詞，可以考慮分批處理，最后union all在一起。

?count distinct 數據傾斜

我們可以修改對應的sql來進行優化，?count+group by?或者sum+group by的方案來優化，在第一階段選出全部的非重復的字段id，在第二階段再對這些已消重的id進行計數

其他優化