數(shù)據(jù)傾斜調(diào)優(yōu)概述

有的時候，我們可能會遇到大數(shù)據(jù)計算中一個最棘手的問題——數(shù)據(jù)傾斜，此時Spark作業(yè)的性能會比期望差很多。數(shù)據(jù)傾斜調(diào)優(yōu)，就是使用各種技術(shù)方案解決不同類型的數(shù)據(jù)傾斜問題，以保證Spark作業(yè)的性能。

數(shù)據(jù)傾斜發(fā)生時的現(xiàn)象

絕大多數(shù)task執(zhí)行得都非?？欤珎€別task執(zhí)行極慢。比如，總共有1000個task，997個task都在1分鐘之內(nèi)執(zhí)行完了，但是剩余兩三個task卻要一兩個小時。這種情況很常見。

原本能夠正常執(zhí)行的Spark作業(yè)，某天突然報出OOM（內(nèi)存溢出）異常，觀察異常棧，是我們寫的業(yè)務(wù)代碼造成的。這種情況比較少見。

數(shù)據(jù)傾斜發(fā)生的原理

數(shù)據(jù)傾斜的原理很簡單：在進行shuffle的時候，必須將各個節(jié)點上相同的key拉取到某個節(jié)點上的一個task來進行處理，比如按照key進行 聚合或join等操作。此時如果某個key對應(yīng)的數(shù)據(jù)量特別大的話，就會發(fā)生數(shù)據(jù)傾斜。比如大部分key對應(yīng)10條數(shù)據(jù)，但是個別key卻對應(yīng)了100萬 條數(shù)據(jù)，那么大部分task可能就只會分配到10條數(shù)據(jù)，然后1秒鐘就運行完了；但是個別task可能分配到了100萬數(shù)據(jù)，要運行一兩個小時。因此，整 個Spark作業(yè)的運行進度是由運行時間最長的那個task決定的。

因此出現(xiàn)數(shù)據(jù)傾斜的時候，Spark作業(yè)看起來會運行得非常緩慢，甚至可能因為某個task處理的數(shù)據(jù)量過大導(dǎo)致內(nèi)存溢出。

下圖就是一個很清晰的例子：hello這個key，在三個節(jié)點上對應(yīng)了總共7條數(shù)據(jù)，這些數(shù)據(jù)都會被拉取到同一個task中進行處理；而world 和you這兩個key分別才對應(yīng)1條數(shù)據(jù)，所以另外兩個task只要分別處理1條數(shù)據(jù)即可。此時第一個task的運行時間可能是另外兩個task的7倍， 而整個stage的運行速度也由運行最慢的那個task所決定。

如何定位導(dǎo)致數(shù)據(jù)傾斜的代碼

數(shù)據(jù)傾斜只會發(fā)生在shuffle過程中。這里給大家羅列一些常用的并且可能會觸發(fā)shuffle操作的算子：distinct、 groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。出現(xiàn)數(shù)據(jù)傾斜時， 可能就是你的代碼中使用了這些算子中的某一個所導(dǎo)致的。

#某個task執(zhí)行特別慢的情況

首先要看的，就是數(shù)據(jù)傾斜發(fā)生在第幾個stage中。

如果是用yarn-client模式提交，那么本地是直接可以看到log的，可以在log中找到當(dāng)前運行到了第幾個stage；如果是用yarn- cluster模式提交，則可以通過Spark Web UI來查看當(dāng)前運行到了第幾個stage。此外，無論是使用yarn-client模式還是yarn-cluster模式，我們都可以在Spark Web UI上深入看一下當(dāng)前這個stage各個task分配的數(shù)據(jù)量，從而進一步確定是不是task分配的數(shù)據(jù)不均勻?qū)е铝藬?shù)據(jù)傾斜。

比如下圖中，倒數(shù)第三列顯示了每個task的運行時間。明顯可以看到，有的task運行特別快，只需要幾秒鐘就可以運行完；而有的task運行特別 慢，需要幾分鐘才能運行完，此時單從運行時間上看就已經(jīng)能夠確定發(fā)生數(shù)據(jù)傾斜了。此外，倒數(shù)第一列顯示了每個task處理的數(shù)據(jù)量，明顯可以看到，運行時 間特別短的task只需要處理幾百KB的數(shù)據(jù)即可，而運行時間特別長的task需要處理幾千KB的數(shù)據(jù)，處理的數(shù)據(jù)量差了10倍。此時更加能夠確定是發(fā)生 了數(shù)據(jù)傾斜。

知道數(shù)據(jù)傾斜發(fā)生在哪一個stage之后，接著我們就需要根據(jù)stage劃分原理，推算出來發(fā)生傾斜的那個stage對應(yīng)代碼中的哪一部分，這部分 代碼中肯定會有一個shuffle類算子。精準(zhǔn)推算stage與代碼的對應(yīng)關(guān)系，需要對Spark的源碼有深入的理解，這里我們可以介紹一個相對簡單實用 的推算方法：只要看到Spark代碼中出現(xiàn)了一個shuffle類算子或者是Spark SQL的SQL語句中出現(xiàn)了會導(dǎo)致shuffle的語句（比如group by語句），那么就可以判定，以那個地方為界限劃分出了前后兩個stage。

這里我們就以Spark最基礎(chǔ)的入門程序——單詞計數(shù)來舉例，如何用最簡單的方法大致推算出一個stage對應(yīng)的代碼。如下示例，在整個代碼中，只 有一個reduceByKey是會發(fā)生shuffle的算子，因此就可以認(rèn)為，以這個算子為界限，會劃分出前后兩個stage。

stage0，主要是執(zhí)行從textFile到map操作，以及執(zhí)行shuffle write操作。shuffle write操作，我們可以簡單理解為對pairs RDD中的數(shù)據(jù)進行分區(qū)操作，每個task處理的數(shù)據(jù)中，相同的key會寫入同一個磁盤文件內(nèi)。

stage1，主要是執(zhí)行從reduceByKey到collect操作，stage1的各個task一開始運行，就會首先執(zhí)行shuffle read操作。執(zhí)行shuffle read操作的task，會從stage0的各個task所在節(jié)點拉取屬于自己處理的那些key，然后對同一個key進行全局性的聚合或join等操作， 在這里就是對key的value值進行累加。stage1在執(zhí)行完reduceByKey算子之后，就計算出了最終的wordCounts RDD，然后會執(zhí)行collect算子，將所有數(shù)據(jù)拉取到Driver上，供我們遍歷和打印輸出。

val conf = new SparkConf()
val sc = new SparkContext(conf)
val lines = sc.textFile("hdfs://...")
val words = lines.flatMap(_.split(" "))
val pairs = words.map((_, 1))
val wordCounts = pairs.reduceByKey(_ + _)
wordCounts.collect().foreach(println(_))

通過對單詞計數(shù)程序的分析，希望能夠讓大家了解最基本的stage劃分的原理，以及stage劃分后shuffle操作 是如何在兩個stage的邊界處執(zhí)行的。然后我們就知道如何快速定位出發(fā)生數(shù)據(jù)傾斜的stage對應(yīng)代碼的哪一個部分了。比如我們在Spark Web UI或者本地log中發(fā)現(xiàn)，stage1的某幾個task執(zhí)行得特別慢，判定stage1出現(xiàn)了數(shù)據(jù)傾斜，那么就可以回到代碼中定位出stage1主要包 括了reduceByKey這個shuffle類算子，此時基本就可以確定是由reduceByKey算子導(dǎo)致的數(shù)據(jù)傾斜問題。比如某個單詞出現(xiàn)了100萬 次，其他單詞才出現(xiàn)10次，那么stage1的某個task就要處理100萬數(shù)據(jù)，整個stage的速度就會被這個task拖慢。

#某個task莫名其妙內(nèi)存溢出的情況

這種情況下去定位出問題的代碼就比較容易了。我們建議直接看yarn-client模式下本地log的異常棧，或者是通過YARN查看yarn- cluster模式下的log中的異常棧。一般來說，通過異常棧信息就可以定位到你的代碼中哪一行發(fā)生了內(nèi)存溢出。然后在那行代碼附近找找，一般也會有 shuffle類算子，此時很可能就是這個算子導(dǎo)致了數(shù)據(jù)傾斜。

但是大家要注意的是，不能單純靠偶然的內(nèi)存溢出就判定發(fā)生了數(shù)據(jù)傾斜。因為自己編寫的代碼的bug，以及偶然出現(xiàn)的數(shù)據(jù)異常，也可能會導(dǎo)致內(nèi)存溢 出。因此還是要按照上面所講的方法，通過Spark Web UI查看報錯的那個stage的各個task的運行時間以及分配的數(shù)據(jù)量，才能確定是否是由于數(shù)據(jù)傾斜才導(dǎo)致了這次內(nèi)存溢出。

#查看導(dǎo)致數(shù)據(jù)傾斜的key的數(shù)據(jù)分布情況

知道了數(shù)據(jù)傾斜發(fā)生在哪里之后，通常需要分析一下那個執(zhí)行了shuffle操作并且導(dǎo)致了數(shù)據(jù)傾斜的RDD/Hive表，查看一下其中key的分布 情況。這主要是為之后選擇哪一種技術(shù)方案提供依據(jù)。針對不同的key分布與不同的shuffle算子組合起來的各種情況，可能需要選擇不同的技術(shù)方案來解決。

#此時根據(jù)你執(zhí)行操作的情況不同，可以有很多種查看key分布的方式：

如果是Spark SQL中的group by、join語句導(dǎo)致的數(shù)據(jù)傾斜，那么就查詢一下SQL中使用的表的key分布情況。

如果是對Spark RDD執(zhí)行shuffle算子導(dǎo)致的數(shù)據(jù)傾斜，那么可以在Spark作業(yè)中加入查看key分布的代碼，比如RDD.countByKey()。然后對統(tǒng)計 出來的各個key出現(xiàn)的次數(shù)，collect/take到客戶端打印一下，就可以看到key的分布情況。

舉例來說，對于上面所說的單詞計數(shù)程序，如果確定了是stage1的reduceByKey算子導(dǎo)致了數(shù)據(jù)傾斜，那么就應(yīng)該看看進行 reduceByKey操作的RDD中的key分布情況，在這個例子中指的就是pairs RDD。如下示例，我們可以先對pairs采樣10%的樣本數(shù)據(jù)，然后使用countByKey算子統(tǒng)計出每個key出現(xiàn)的次數(shù)，最后在客戶端遍歷和打印 樣本數(shù)據(jù)中各個key的出現(xiàn)次數(shù)。

val sampledPairs = pairs.sample(false, 0.1)
val sampledWordCounts = sampledPairs.countByKey()
sampledWordCounts.foreach(println(_))

#