一. 資料傾斜的現象

多數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上，所以資料量不能太大。

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