Spark性能調優之解決數據傾斜
數據傾斜七種解決方案
shuffle的過程最容易引起數據傾斜
1.使用Hive ETL預處理數據
• 方案適用場景：如果導致數據傾斜的是Hive表。如果該Hive表中的數據本身很不均勻（比如某個
key對應了100萬數據，其他key才對應了10條數據），而且業務場景需要頻繁使用Spark對Hive表
執行某個分析操作，那么比較適合使用這種技術方案。
• 方案實現思路：此時可以評估一下，是否可以通過Hive來進行數據預處理（即通過Hive ETL預先對
數據按照key進行聚合，或者是預先和其他表進行join），然后在Spark作業中針對的數據源就不是
原來的Hive表了，而是預處理后的Hive表。此時由於數據已經預先進行過聚合或join操作了，那么
在Spark作業中也就不需要使用原先的shuffle類算子執行這類操作了。
• 方案實現原理：這種方案從根源上解決了數據傾斜，因為徹底避免了在Spark中執行shuffle類算子
，那么肯定就不會有數據傾斜的問題了。但是這里也要提醒一下大家，這種方式屬於治標不治本。
因為畢竟數據本身就存在分布不均勻的問題，所以Hive ETL中進行group by或者join等shuffle操作
時，還是會出現數據傾斜，導致Hive ETL的速度很慢。我們只是把數據傾斜的發生提前到了Hive
ETL中，避免Spark程序發生數據傾斜而已。
2.過濾少數導致傾斜的key
SampleOperator
• 方案適用場景：如果發現導致傾斜的key就少數幾個，而且對計算本身的影響並不大的話，那么很
適合使用這種方案。比如99%的key就對應10條數據，但是只有一個key對應了100萬數據，從而導
致了數據傾斜。
• 方案實現思路：如果我們判斷那少數幾個數據量特別多的key，對作業的執行和計算結果不是特別
重要的話，那么干脆就直接過濾掉那少數幾個key。比如，在Spark SQL中可以使用where子句過濾
掉這些key或者在Spark Core中對RDD執行filter算子過濾掉這些key。如果需要每次作業執行時，
動態判定哪些key的數據量最多然后再進行過濾，那么可以使用sample算子對RDD進行采樣，然后
計算出每個key的數量，取數據量最多的key過濾掉即可。
• 方案實現原理：將導致數據傾斜的key給過濾掉之后，這些key就不會參與計算了，自然不可能產生
數據傾斜。
sample算子的使用
nameRDD.sample(false,0.75)
false代表不放回的抽樣，0.75代表從整體中抽多少數據
3.提高shuffle操作的並行度
• 方案實現思路：在對RDD執行shuffle算子時，給shuffle算子傳入一個參數，
比如
reduceByKey(1000)，該參數就設置了這個shuffle算子執行時shuffle read task的數量。對於
Spark SQL中的shuffle類語句，比如group by、join等，需要設置一個參數，即
spark.sql.shuffle.partitions，該參數代表了shuffle read task的並行度，該值默認是200，對於很
多場景來說都有點過小。
• 方案實現原理：增加shuffle read task的數量，可以讓原本分配給一個task的多個key分配給多個
task，從而讓每個task處理比原來更少的數據。舉例來說，如果原本有5個key，每個key對應10條
數據，這5個key都是分配給一個task的，那么這個task就要處理50條數據。而增加了shuffle read
task以后，每個task就分配到一個key，即每個task就處理10條數據，那么自然每個task的執行時
間都會變短了。具體原理如下圖所示
4.雙重聚合
DoubelReduceByKey
• 方案適用場景：對RDD執行reduceByKey等聚合類shuffle算子或者在Spark SQL中使用group by
語句進行分組聚合時，比較適用這種方案。
• 方案實現思路：這個方案的核心實現思路就是進行兩階段聚合。第一次是局部聚合，先給每個key
都打上一個隨機數，比如10以內的隨機數，此時原先一樣的key就變成不一樣的了，比如(hello, 1)
(hello, 1) (hello, 1) (hello, 1)，就會變成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接着
對打上隨機數后的數據，執行reduceByKey等聚合操作，進行局部聚合，那么局部聚合結果，就會
變成了(1_hello, 2) (2_hello, 2)。然后將各個key的前綴給去掉，就會變成(hello,2)(hello,2)，再次
進行全局聚合操作，就可以得到最終結果了，比如(hello, 4)。
• 方案實現原理：將原本相同的key通過附加隨機前綴的方式，變成多個不同的key，就可以讓原本被
一個task處理的數據分散到多個task上去做局部聚合，進而解決單個task處理數據量過多的問題。
接着去除掉隨機前綴，再次進行全局聚合，就可以得到最終的結果
5.將reduce join轉為map join
• 方案適用場景：在對RDD使用join類操作，或者是在Spark SQL中使用join語句時，而且join操作中
的一個RDD或表的數據量比較小（比如幾百M或者一兩G），
具體看Executor的內存來廣播
（executor-memory * 0.48 0.6 * 0.8），比較適用此方案。
• 方案實現思路：不使用join算子進行連接操作，而使用Broadcast變量與map類算子實現join操作，
進而完全規避掉shuffle類的操作，徹底避免數據傾斜的發生和出現。將較小RDD中的數據直接通過
collect算子拉取到Driver端的內存中來，然后對其創建一個Broadcast變量；接着對另外一個RDD
執行map類算子，在算子函數內，從Broadcast變量中獲取較小RDD的全量數據，與當前RDD的每
一條數據按照連接key進行比對，如果連接key相同的話，那么就將兩個RDD的數據用你需要的方式
連接起來。
• 方案實現原理：普通的join是會走shuffle過程的，而一旦shuffle，就相當於會將相同key的數據拉
取到一個shuffle read task中再進行join，此時就是reduce join。但是如果一個RDD是比較小的，
則可以采用
廣播小RDD全量數據+map算子來實現與join同樣的效果，也就是map join，此時就不
會發生shuffle操作，也就不會發生數據傾斜
6.采樣傾斜key並分拆join操作
DoubleJoin
• 方案適用場景：兩個RDD/Hive表進行join的時候，如果數據量都比較大，無法采用“解決方案五
”，那么此時可以看一下兩個RDD/Hive表中的key分布情況。
如果出現數據傾斜，是因為其中某一
個RDD/Hive表中的少數幾個key的數據量過大，而另一個RDD/Hive表中的所有key都分布比較均
勻，那么采用這個解決方案是比較合適的。
• 方案實現思路：
– 對包含少數幾個數據量過大的key的那個RDD，通過sample算子采樣出一份樣本來，然后統計一下每個
key的數量，計算出來數據量最大的是哪幾個key。
– 然后將這幾個key對應的數據從原來的RDD中拆分出來，形成一個單獨的RDD，並給每個key都打上n以
內的隨機數作為前綴，而不會導致傾斜的大部分key形成另外一個RDD。
– 接着將需要join的另一個RDD，也過濾出來那幾個傾斜key對應的數據並形成一個單獨的RDD，將每條數
據膨脹成n條數據，這n條數據都按順序附加一個0~n的前綴，不會導致傾斜的大部分key也形成另外一個
RDD。
– 再將附加了隨機前綴的獨立RDD與另一個膨脹n倍的獨立RDD進行join，此時就可以將原先相同的key打
散成n份，分散到多個task中去進行join了。
– 而另外兩個普通的RDD就照常join即可。
– 最后將兩次join的結果使用union算子合並起來即可，就是最終的join結果。
7.使用隨機前綴和擴容RDD進行join
• 方案適用場景：如果在進行join操作時，RDD中有大量的key導致數據傾斜，那么進行分拆key也沒
什么意義，此時就只能使用最后一種方案來解決問題了。
• 方案實現思路：
– 該方案的實現思路基本和“解決方案六”類似，首先查看RDD/Hive表中的數據分布情況，找到那個造成
數據傾斜的RDD/Hive表，比如有多個key都對應了超過1萬條數據。
– 然后將該RDD的每條數據都打上一個n以內的隨機前綴。
– 同時對另外一個正常的RDD進行擴容，將每條數據都擴容成n條數據，擴容出來的每條數據都依次打上一
個0~n的前綴。
– 最后將兩個處理后的RDD進行join即可。
• 方案實現原理：將原先一樣的key通過附加隨機前綴變成不一樣的key，然后就可以將這些處理后的
“不同key”分散到多個task中去處理，而不是讓一個task處理大量的相同key。該方案與“解決方
案六”的不同之處就在於，上一種方案是盡量只對少數傾斜key對應的數據進行特殊處理，由於處
理過程需要擴容RDD，因此上一種方案擴容RDD后對內存的占用並不大；而這一種方案是針對有大
量傾斜key的情況，沒法將部分key拆分出來進行單獨處理，因此只能對整個RDD進行數據擴容，對
內存資源要求很高。