一. 數據傾斜的現象

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

二. 數據傾斜的原因

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

數據問題

key本身分布不均勻(包括大量的key為空)

key的設置不合理

spark使用問題

shuffle時的并發度不夠

計算方式有誤

三. 數據傾斜的后果

spark中一個stage的執行時間受限于最后那個執行完的task，因此運行緩慢的任務會拖累整個程序的運行速度（分布式程序運行的速度是由最慢的那個task決定的）。

過多的數據在同一個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)

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

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

null（空值）或是一些無意義的信息()之類的,大多是這個原因引起。

無效數據，大量重復的測試數據或是對結果影響不大的有效數據。

有效數據，業務導致的正常數據分布。

解決辦法

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

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

隔離執行，將異常的key過濾出來單獨處理，最后與正常數據的處理結果進行union操作。

對key先添加隨機值，進行操作后，去掉隨機值，再進行一次操作。

使用reduceByKey?代替?groupByKey

使用map join。

舉例：

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

將原始的?key?轉化為?key + 隨機值(例如Random.nextInt)

對數據進行?reduceByKey(func)

將?key + 隨機值?轉成?key

再對數據進行?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過程中數據傾斜的處理

過濾無效數據，如空值、測試數據等等

在map端使用combiner函數

局部聚合加全局聚合。

先對key加隨機后綴，然后進行reduce操作

對第一次執行的結果再此進行MR操作。（在map端去掉后綴后再進行reduce操作）

?

轉載于:https://www.cnblogs.com/0xcafedaddy/p/7610613.html

總結

以上是生活随笔為你收集整理的解决spark中遇到的数据倾斜问题的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。