解決spark中遇到的數據傾斜問題

一. 數據傾斜的現象

多數task執行速度較快,少數task執行時間非常長，或者等待很長時間后提示你內存不足，執行失敗。

二. 數據傾斜的原因

常見於各種shuffle操作，例如reduceByKey,groupByKey,join等操作。

數據問題

1. key本身分布不均勻(包括大量的key為空)
2. key的設置不合理

spark使用問題

1. shuffle時的並發度不夠
2. 計算方式有誤

三. 數據傾斜的后果

1. spark中一個stage的執行時間受限於最后那個執行完的task，因此運行緩慢的任務會拖累整個程序的運行速度（分布式程序運行的速度是由最慢的那個task決定的）。
2. 過多的數據在同一個task中執行，將會把executor撐爆，造成OOM，程序終止運行。

一個理想的分布式程序： 

發生數據傾斜時，任務的執行速度由最大的那個任務決定： 

四. 數據問題造成的數據傾斜

發現數據傾斜的時候，不要急於提高executor的資源，修改參數或是修改程序，首先要檢查數據本身，是否存在異常數據。

找出異常的key

如果任務長時間卡在最后最后1個(幾個)任務，首先要對key進行抽樣分析，判斷是哪些key造成的。

選取key，對數據進行抽樣，統計出現的次數，根據出現次數大小排序取出前幾個

df.select("key").sample(false,0.1).(k=>(k,1)).reduceBykey(_+_).map(k=>(k._2,k._1)).sortByKey(false).take(10)

如果發現多數數據分布都較為平均，而個別數據比其他數據大上若干個數量級，則說明發生了數據傾斜。

經過分析，傾斜的數據主要有以下三種情況:

1. null（空值）或是一些無意義的信息()之類的,大多是這個原因引起。
2. 無效數據，大量重復的測試數據或是對結果影響不大的有效數據。
3. 有效數據，業務導致的正常數據分布。

解決辦法

第1，2種情況，直接對數據進行過濾即可。

第3種情況則需要進行一些特殊操作，常見的有以下幾種做法。

1. 隔離執行，將異常的key過濾出來單獨處理，最后與正常數據的處理結果進行union操作。
2. 對key先添加隨機值，進行操作后，去掉隨機值，再進行一次操作。
3. 使用reduceByKey 代替 groupByKey
4. 使用map join。

舉例：

如果使用reduceByKey因為數據傾斜造成運行失敗的問題。具體操作如下：

1. 將原始的 key 轉化為 key + 隨機值(例如Random.nextInt)
2. 對數據進行 reduceByKey(func)
3. 將 key + 隨機值 轉成 key
4. 再對數據進行 reduceByKey(func)

tip1: 如果此時依舊存在問題，建議篩選出傾斜的數據單獨處理。最后將這份數據與正常的數據進行union即可。

tips2: 單獨處理異常數據時，可以配合使用Map Join解決。

五. spark使用不當造成的數據傾斜

1. 提高shuffle並行度

dataFrame和sparkSql可以設置spark.sql.shuffle.partitions參數控制shuffle的並發度，默認為200。 
rdd操作可以設置spark.default.parallelism控制並發度，默認參數由不同的Cluster Manager控制。

局限性: 只是讓每個task執行更少的不同的key。無法解決個別key特別大的情況造成的傾斜，如果某些key的大小非常大，即使一個task單獨執行它，也會受到數據傾斜的困擾。

2. 使用map join 代替reduce join

在小表不是特別大(取決於你的executor大小)的情況下使用，可以使程序避免shuffle的過程，自然也就沒有數據傾斜的困擾了。

局限性: 因為是先將小數據發送到每個executor上，所以數據量不能太大。

具體使用方法和處理流程參照:

Spark map-side-join 關聯優化

spark join broadcast優化

六. MapReduce過程中數據傾斜的處理

1. 過濾無效數據，如空值、測試數據等等
2. 在map端使用combiner函數
3. 局部聚合加全局聚合。
   1. 先對key加隨機后綴，然后進行reduce操作
   2. 對第一次執行的結果再此進行MR操作。（在map端去掉后綴后再進行reduce操作）