[[201369]]

本文將結(jié)合實例詳細(xì)闡明Spark數(shù)據(jù)傾斜(Data Skew)的幾種場景及對應(yīng)的解決方案，包括避免數(shù)據(jù)源傾斜、調(diào)整并行度、使用自定義Partitioner、使用Map側(cè)Join代替Reduce側(cè)Join、給傾斜Key加上隨機前綴等。

一、為何要處理數(shù)據(jù)傾斜

1、什么是數(shù)據(jù)傾斜

對Spark/Hadoop這樣的大數(shù)據(jù)系統(tǒng)來講，數(shù)據(jù)量大并不可怕，可怕的是數(shù)據(jù)傾斜。

那何謂數(shù)據(jù)傾斜?數(shù)據(jù)傾斜指的是并行處理的數(shù)據(jù)集中，某一部分(如Spark或Kafka的一個Partition)的數(shù)據(jù)顯著多于其它部分，從而使得該部分的處理速度成為整個數(shù)據(jù)集處理的瓶頸。

2、數(shù)據(jù)傾斜是如何造成的

在Spark中，同一個Stage的不同Partition可以并行處理，而具有依賴關(guān)系的不同Stage之間是串行處理的。假設(shè)某個Spark Job分為Stage 0和Stage 1兩個Stage，且Stage 1依賴于Stage 0，那Stage 0完全處理結(jié)束之前不會處理Stage 1。而Stage 0可能包含N個Task，這N個Task可以并行進(jìn)行。如果其中N-1個Task都在10秒內(nèi)完成，而另外一個Task卻耗時1分鐘，那該Stage的總時間至少為1分鐘。換句話說，一個Stage所耗費的時間，主要由最慢的那個Task決定。

由于同一個Stage內(nèi)的所有Task執(zhí)行相同的計算，在排除不同計算節(jié)點計算能力差異的前提下，不同Task之間耗時的差異主要由該Task所處理的數(shù)據(jù)量決定。

Stage的數(shù)據(jù)來源主要分為如下兩類：

• 從數(shù)據(jù)源直接讀取，如讀取HDFS、Kafka
• 讀取上一個Stage的Shuffle數(shù)據(jù)

二、如何緩解/消除數(shù)據(jù)傾斜

1、盡量避免數(shù)據(jù)源的數(shù)據(jù)傾斜

以Spark Stream通過DirectStream方式讀取Kafka數(shù)據(jù)為例。由于Kafka的每一個Partition對應(yīng)Spark的一個Task(Partition)，所以Kafka內(nèi)相關(guān)Topic的各Partition之間數(shù)據(jù)是否平衡，直接決定Spark處理該數(shù)據(jù)時是否會產(chǎn)生數(shù)據(jù)傾斜。

Kafka某一Topic內(nèi)消息在不同Partition之間的分布，主要由Producer端所使用的Partition實現(xiàn)類決定。如果使用隨機Partitioner，則每條消息會隨機發(fā)送到一個Partition中，那么從概率上來講，各Partition間的數(shù)據(jù)會達(dá)到平衡。此時源Stage(直接讀取Kafka數(shù)據(jù)的Stage)不會產(chǎn)生數(shù)據(jù)傾斜。

但很多時候，業(yè)務(wù)場景可能會要求將具備同一特征的數(shù)據(jù)順序消費，此時就需要將具有相同特征的數(shù)據(jù)放于同一個Partition中。一個典型的場景是，需要將同一個用戶相關(guān)的PV信息置于同一個Partition中。此時，如果產(chǎn)生了數(shù)據(jù)傾斜，則需要通過其它方式處理。

2、調(diào)整并行度分散同一個Task的不同Key

原理

Spark在做Shuffle時，默認(rèn)使用HashPartitioner(非Hash Shuffle)對數(shù)據(jù)進(jìn)行分區(qū)。如果并行度設(shè)置得不合適，可能造成大量不相同的Key對應(yīng)的數(shù)據(jù)被分配到了同一個Task上，造成該Task所處理的數(shù)據(jù)遠(yuǎn)大于其它Task，從而造成數(shù)據(jù)傾斜。

如果調(diào)整Shuffle時的并行度，使得原本被分配到同一Task的不同Key發(fā)配到不同Task上處理，則可降低原Task所需處理的數(shù)據(jù)量，從而緩解數(shù)據(jù)傾斜問題造成的短板效應(yīng)

案例

現(xiàn)有一張測試表，名為student_external，內(nèi)有10.5億條數(shù)據(jù)，每條數(shù)據(jù)有一個唯一的id值?，F(xiàn)從中取出id取值為9億到10.5億的共1.5億條數(shù)據(jù)，并通過一些處理，使得id為9億到9.4億間的所有數(shù)據(jù)對12取模后余數(shù)為8(即在Shuffle并行度為12時該數(shù)據(jù)集全部被HashPartition分配到第8個Task)，其它數(shù)據(jù)集對其id除以100取整，從而使得id大于9.4億的數(shù)據(jù)在Shuffle時可被均勻分配到所有Task中，而id小于9.4億的數(shù)據(jù)全部分配到同一個Task中。

處理過程如下：

通過上述處理，一份可能造成后續(xù)數(shù)據(jù)傾斜的測試數(shù)據(jù)即以準(zhǔn)備好。接下來，使用Spark讀取該測試數(shù)據(jù)，并通過groupByKey(12)對id分組處理，且Shuffle并行度為12。代碼如下：

本次實驗所使用集群節(jié)點數(shù)為4，每個節(jié)點可被Yarn使用的CPU核數(shù)為16，內(nèi)存為16GB。使用如下方式提交上述應(yīng)用，將啟動4個Executor，每個Executor可使用核數(shù)為12(該配置并非生產(chǎn)環(huán)境下的***配置，僅用于本文實驗)，可用內(nèi)存為12GB。

spark-submit --queue ambari --num-executors 4 --executor-cores 12 --executor-memory 12g --class com.jasongj.spark.driver.SparkDataSkew --master yarn --deploy-mode client SparkExample-with-dependencies-1.0.jar 

GroupBy Stage的Task狀態(tài)如下圖所示，Task 8處理的記錄數(shù)為4500萬，遠(yuǎn)大于(9倍于)其它11個Task處理的500萬記錄。而Task 8所耗費的時間為38秒，遠(yuǎn)高于其它11個Task的平均時間(16秒)。整個Stage的時間也為38秒，該時間主要由最慢的Task 8決定。

在這種情況下，可以通過調(diào)整Shuffle并行度，使得原來被分配到同一個Task(即該例中的Task 8)的不同Key分配到不同Task，從而降低Task 8所需處理的數(shù)據(jù)量，緩解數(shù)據(jù)傾斜。

通過groupByKey(48)將Shuffle并行度調(diào)整為48，重新提交到Spark。新的Job的GroupBy Stage所有Task狀態(tài)如下圖所示。

從上圖可知，記錄數(shù)最多的Task 20處理的記錄數(shù)約為1125萬，相比于并行度為12時Task 8的4500萬，降低了75%左右，而其耗時從原來Task 8的38秒降到了24秒。

在這種場景下，調(diào)整并行度，并不意味著一定要增加并行度，也可能是減小并行度。如果通過groupByKey(11)將Shuffle并行度調(diào)整為11，重新提交到Spark。新Job的GroupBy Stage的所有Task狀態(tài)如下圖所示。

從上圖可見，處理記錄數(shù)最多的Task 6所處理的記錄數(shù)約為1045萬，耗時為23秒。處理記錄數(shù)最少的Task 1處理的記錄數(shù)約為545萬，耗時12秒。

小結(jié)

適用場景：

大量不同的Key被分配到了相同的Task造成該Task數(shù)據(jù)量過大。

解決方案：

調(diào)整并行度。一般是增大并行度，但有時如本例減小并行度也可達(dá)到效果。

優(yōu)勢：

實現(xiàn)簡單，可在需要Shuffle的操作算子上直接設(shè)置并行度或者使用spark.default.parallelism設(shè)置。如果是Spark SQL，還可通過SET spark.sql.shuffle.partitions=[num_tasks]設(shè)置并行度。可用最小的代價解決問題。一般如果出現(xiàn)數(shù)據(jù)傾斜，都可以通過這種方法先試驗幾次，如果問題未解決，再嘗試其它方法。

劣勢：

適用場景少，只能將分配到同一Task的不同Key分散開，但對于同一Key傾斜嚴(yán)重的情況該方法并不適用。并且該方法一般只能緩解數(shù)據(jù)傾斜，沒有徹底消除問題。從實踐經(jīng)驗來看，其效果一般。

3、自定義Partitioner

原理

使用自定義的Partitioner(默認(rèn)為HashPartitioner)，將原本被分配到同一個Task的不同Key分配到不同Task。

案例

以上述數(shù)據(jù)集為例，繼續(xù)將并發(fā)度設(shè)置為12，但是在groupByKey算子上，使用自定義的Partitioner(實現(xiàn)如下)：

由下圖可見，使用自定義Partition后，耗時最長的Task 6處理約1000萬條數(shù)據(jù)，用時15秒。并且各Task所處理的數(shù)據(jù)集大小相當(dāng)。

小結(jié)

適用場景：

大量不同的Key被分配到了相同的Task，造成該Task數(shù)據(jù)量過大。

解決方案：

使用自定義的Partitioner實現(xiàn)類代替默認(rèn)的HashPartitioner，盡量將所有不同的Key均勻分配到不同的Task中。

優(yōu)勢：

不影響原有的并行度設(shè)計。如果改變并行度，后續(xù)Stage的并行度也會默認(rèn)改變，可能會影響后續(xù)Stage。

劣勢：

適用場景有限，只能將不同Key分散開，對于同一Key對應(yīng)數(shù)據(jù)集非常大的場景不適用。效果與調(diào)整并行度類似，只能緩解數(shù)據(jù)傾斜而不能完全消除數(shù)據(jù)傾斜。而且需要根據(jù)數(shù)據(jù)特點自定義專用的Partitioner，不夠靈活。

4、將Reduce side Join轉(zhuǎn)變?yōu)镸ap side Join

原理

通過Spark的Broadcast機制，將Reduce側(cè)Join轉(zhuǎn)化為Map側(cè)Join，避免Shuffle從而完全消除Shuffle帶來的數(shù)據(jù)傾斜。

案例

通過如下SQL創(chuàng)建一張具有傾斜Key且總記錄數(shù)為1.5億的大表test。

使用如下SQL創(chuàng)建一張數(shù)據(jù)分布均勻且總記錄數(shù)為50萬的小表test_new。

直接通過Spark Thrift Server提交如下SQL將表test與表test_new進(jìn)行Join并將Join結(jié)果存于表test_join中。

該SQL對應(yīng)的DAG如下圖所示。從該圖可見，該執(zhí)行過程總共分為三個Stage，前兩個用于從Hive中讀取數(shù)據(jù)，同時二者進(jìn)行Shuffle，通過***一個Stage進(jìn)行Join并將結(jié)果寫入表test_join中。

從下圖可見，Join Stage各Task處理的數(shù)據(jù)傾斜嚴(yán)重，處理數(shù)據(jù)量***的Task耗時7.1分鐘，遠(yuǎn)高于其它無數(shù)據(jù)傾斜的Task約2秒的耗時。

接下來，嘗試通過Broadcast實現(xiàn)Map側(cè)Join。實現(xiàn)Map側(cè)Join的方法，并非直接通過CACHE TABLE test_new將小表test_new進(jìn)行cache?，F(xiàn)通過如下SQL進(jìn)行Join。

通過如下DAG圖可見，該操作仍分為三個Stage，且仍然有Shuffle存在，唯一不同的是，小表的讀取不再直接掃描Hive表，而是掃描內(nèi)存中緩存的表。

并且數(shù)據(jù)傾斜仍然存在。如下圖所示，最慢的Task耗時為7.1分鐘，遠(yuǎn)高于其它Task的約2秒。

正確的使用Broadcast實現(xiàn)Map側(cè)Join的方式是，通過SET spark.sql.autoBroadcastJoinThreshold=104857600;將Broadcast的閾值設(shè)置得足夠大。

再次通過如下SQL進(jìn)行Join。

通過如下DAG圖可見，該方案只包含一個Stage。

并且從下圖可見，各Task耗時相當(dāng)，無明顯數(shù)據(jù)傾斜現(xiàn)象。并且總耗時為1.5分鐘，遠(yuǎn)低于Reduce側(cè)Join的7.3分鐘。

小結(jié)

適用場景：

參與Join的一邊數(shù)據(jù)集足夠小，可被加載進(jìn)Driver并通過Broadcast方法廣播到各個Executor中。

解決方案：

在Java/Scala代碼中將小數(shù)據(jù)集數(shù)據(jù)拉取到Driver，然后通過Broadcast方案將小數(shù)據(jù)集的數(shù)據(jù)廣播到各Executor。或者在使用SQL前，將Broadcast的閾值調(diào)整得足夠多，從而使用Broadcast生效。進(jìn)而將Reduce側(cè)Join替換為Map側(cè)Join。

優(yōu)勢：

避免了Shuffle，徹底消除了數(shù)據(jù)傾斜產(chǎn)生的條件，可極大提升性能。

劣勢：

要求參與Join的一側(cè)數(shù)據(jù)集足夠小，并且主要適用于Join的場景，不適合聚合的場景，適用條件有限。

5、為skew的key增加隨機前/后綴

原理

為數(shù)據(jù)量特別大的Key增加隨機前/后綴，使得原來Key相同的數(shù)據(jù)變?yōu)镵ey不相同的數(shù)據(jù)，從而使傾斜的數(shù)據(jù)集分散到不同的Task中，徹底解決數(shù)據(jù)傾斜問題。Join另一則的數(shù)據(jù)中，與傾斜Key對應(yīng)的部分?jǐn)?shù)據(jù)，與隨機前綴集作笛卡爾乘積，從而保證無論數(shù)據(jù)傾斜側(cè)傾斜Key如何加前綴，都能與之正常Join。

案例

通過如下SQL，將id為9億到9.08億共800萬條數(shù)據(jù)的id轉(zhuǎn)為9500048或者9500096，其它數(shù)據(jù)的id除以100取整。從而該數(shù)據(jù)集中，id為9500048和9500096的數(shù)據(jù)各400萬，其它id對應(yīng)的數(shù)據(jù)記錄數(shù)均為100條。這些數(shù)據(jù)存于名為test的表中。

對于另外一張小表test_new，取出50萬條數(shù)據(jù)，并將id(遞增且唯一)除以100取整，使得所有id都對應(yīng)100條數(shù)據(jù)。

通過如下代碼，讀取test表對應(yīng)的文件夾內(nèi)的數(shù)據(jù)并轉(zhuǎn)換為JavaPairRDD存于leftRDD中，同樣讀取test表對應(yīng)的數(shù)據(jù)存于rightRDD中。通過RDD的join算子對leftRDD與rightRDD進(jìn)行Join，并指定并行度為48。

從下圖可看出，整個Join耗時1分54秒，其中Join Stage耗時1.7分鐘。

通過分析Join Stage的所有Task可知，在其它Task所處理記錄數(shù)為192.71萬的同時Task 32的處理的記錄數(shù)為992.72萬，故它耗時為1.7分鐘，遠(yuǎn)高于其它Task的約10秒。這與上文準(zhǔn)備數(shù)據(jù)集時，將id為9500048為9500096對應(yīng)的數(shù)據(jù)量設(shè)置非常大，其它id對應(yīng)的數(shù)據(jù)集非常均勻相符合。

現(xiàn)通過如下操作，實現(xiàn)傾斜Key的分散處理：

將leftRDD中傾斜的key(即9500048與9500096)對應(yīng)的數(shù)據(jù)單獨過濾出來，且加上1到24的隨機前綴，并將前綴與原數(shù)據(jù)用逗號分隔(以方便之后去掉前綴)形成單獨的leftSkewRDD

將rightRDD中傾斜key對應(yīng)的數(shù)據(jù)抽取出來，并通過flatMap操作將該數(shù)據(jù)集中每條數(shù)據(jù)均轉(zhuǎn)換為24條數(shù)據(jù)(每條分別加上1到24的隨機前綴)，形成單獨的rightSkewRDD

將leftSkewRDD與rightSkewRDD進(jìn)行Join，并將并行度設(shè)置為48，且在Join過程中將隨機前綴去掉，得到傾斜數(shù)據(jù)集的Join結(jié)果skewedJoinRDD

將leftRDD中不包含傾斜Key的數(shù)據(jù)抽取出來作為單獨的leftUnSkewRDD

對leftUnSkewRDD與原始的rightRDD進(jìn)行Join，并行度也設(shè)置為48，得到Join結(jié)果unskewedJoinRDD

通過union算子將skewedJoinRDD與unskewedJoinRDD進(jìn)行合并，從而得到完整的Join結(jié)果集

具體實現(xiàn)代碼如下：

從下圖可看出，整個Join耗時58秒，其中Join Stage耗時33秒。

通過分析Join Stage的所有Task可知：

由于Join分傾斜數(shù)據(jù)集Join和非傾斜數(shù)據(jù)集Join，而各Join的并行度均為48，故總的并行度為96

由于提交任務(wù)時，設(shè)置的Executor個數(shù)為4，每個Executor的core數(shù)為12，故可用Core數(shù)為48，所以前48個Task同時啟動(其Launch時間相同)，后48個Task的啟動時間各不相同(等待前面的Task結(jié)束才開始)

由于傾斜Key被加上隨機前綴，原本相同的Key變?yōu)椴煌腒ey，被分散到不同的Task處理，故在所有Task中，未發(fā)現(xiàn)所處理數(shù)據(jù)集明顯高于其它Task的情況

實際上，由于傾斜Key與非傾斜Key的操作完全獨立，可并行進(jìn)行。而本實驗受限于可用總核數(shù)為48，可同時運行的總Task數(shù)為48，故而該方案只是將總耗時減少一半(效率提升一倍)。如果資源充足，可并發(fā)執(zhí)行Task數(shù)增多，該方案的優(yōu)勢將更為明顯。在實際項目中，該方案往往可提升數(shù)倍至10倍的效率。

小結(jié)

適用場景：

兩張表都比較大，無法使用Map則Join。其中一個RDD有少數(shù)幾個Key的數(shù)據(jù)量過大，另外一個RDD的Key分布較為均勻。

解決方案：

將有數(shù)據(jù)傾斜的RDD中傾斜Key對應(yīng)的數(shù)據(jù)集單獨抽取出來加上隨機前綴，另外一個RDD每條數(shù)據(jù)分別與隨機前綴結(jié)合形成新的RDD(相當(dāng)于將其數(shù)據(jù)增到到原來的N倍，N即為隨機前綴的總個數(shù))，然后將二者Join并去掉前綴。然后將不包含傾斜Key的剩余數(shù)據(jù)進(jìn)行Join。***將兩次Join的結(jié)果集通過union合并，即可得到全部Join結(jié)果。

優(yōu)勢：

相對于Map則Join，更能適應(yīng)大數(shù)據(jù)集的Join。如果資源充足，傾斜部分?jǐn)?shù)據(jù)集與非傾斜部分?jǐn)?shù)據(jù)集可并行進(jìn)行，效率提升明顯。且只針對傾斜部分的數(shù)據(jù)做數(shù)據(jù)擴(kuò)展，增加的資源消耗有限。

劣勢：

如果傾斜Key非常多，則另一側(cè)數(shù)據(jù)膨脹非常大，此方案不適用。而且此時對傾斜Key與非傾斜Key分開處理，需要掃描數(shù)據(jù)集兩遍，增加了開銷。

6、大表隨機添加N種隨機前綴，小表擴(kuò)大N倍

原理

如果出現(xiàn)數(shù)據(jù)傾斜的Key比較多，上一種方法將這些大量的傾斜Key分拆出來，意義不大。此時更適合直接對存在數(shù)據(jù)傾斜的數(shù)據(jù)集全部加上隨機前綴，然后對另外一個不存在嚴(yán)重數(shù)據(jù)傾斜的數(shù)據(jù)集整體與隨機前綴集作笛卡爾乘積(即將數(shù)據(jù)量擴(kuò)大N倍)。

案例

這里給出示例代碼，讀者可參考上文中分拆出少數(shù)傾斜Key添加隨機前綴的方法，自行測試。

小結(jié)

適用場景：

一個數(shù)據(jù)集存在的傾斜Key比較多，另外一個數(shù)據(jù)集數(shù)據(jù)分布比較均勻。

優(yōu)勢：

對大部分場景都適用，效果不錯。

劣勢：

需要將一個數(shù)據(jù)集整體擴(kuò)大N倍，會增加資源消耗。

三、總結(jié)

對于數(shù)據(jù)傾斜，并無一個統(tǒng)一的一勞永逸的方法。更多的時候，是結(jié)合數(shù)據(jù)特點(數(shù)據(jù)集大小，傾斜Key的多少等)綜合使用上文所述的多種方法。希望本文能對你有所幫助和啟發(fā)。