Spark性能優(yōu)化(數(shù)據(jù)傾斜調(diào)優(yōu))

1、Spark性能優(yōu)化：數(shù)據(jù)傾斜曹征1.數(shù)據(jù)傾斜曹征調(diào)整簡介在某些情況下，大數(shù)據(jù)計算中最棘手的問題之一數(shù)據(jù)可能會傾斜。此時，火花工作的性能會遠遠低于預(yù)期。調(diào)整數(shù)據(jù)傾斜是通過使用多種技術(shù)解決方案解決不同類型的數(shù)據(jù)傾斜問題，以確?；鸹ú僮鞯男阅?。數(shù)據(jù)傾斜時的現(xiàn)象大多數(shù)task的運行速度非?？?，但單個task的運行速度非常慢。例如，您總共有1000個task，997個task在一分鐘內(nèi)運行，但剩馀的2-3個task需要1-2個小時。這種情況很常見。原來正常運行的火花作業(yè)有一天突然報告內(nèi)存溢出(OOM)異常，觀察異常堆棧是因為我們編寫的業(yè)務(wù)代碼。這種情況比較罕見。數(shù)據(jù)傾斜的發(fā)生原理數(shù)據(jù)傾斜的原理很簡單。在

2、shuffle過程中，必須將每個節(jié)點上的相同關(guān)鍵點拖動到節(jié)點上的一個task中進行處理。例如，按key聚集或執(zhí)行join。此時，如果特定鍵的數(shù)據(jù)楊怡特別多，則會發(fā)生數(shù)據(jù)傾斜。例如，大部分密鑰對應(yīng)于10個數(shù)據(jù)，而單個密鑰對應(yīng)于100萬個數(shù)據(jù)，則大部分task只能分配給10個數(shù)據(jù)，并且可以在一秒內(nèi)運行。但是，單個task可能分配給運行1到2個小時的100萬個數(shù)據(jù)。因此，整個Spark作業(yè)的執(zhí)行進度由運行時間最長的相應(yīng)task確定。因此，如果數(shù)據(jù)傾斜，Spark操作看起來運行得很慢，甚至可能因特定task處理的數(shù)據(jù)量過多而發(fā)生內(nèi)存泄漏。下圖是一個明顯的示例。名為hello的key在三個節(jié)點上共對應(yīng)于

3、7個數(shù)據(jù)，所有這些數(shù)據(jù)都被拉至同一task進行處理。World和you分別對應(yīng)于一個數(shù)據(jù)，因此其他兩個task分別處理一個數(shù)據(jù)就可以了。第一個task的運行時間是其他兩個task的7倍，并且整個stage的運行速度由最慢的task決定。如何查找導(dǎo)致數(shù)據(jù)傾斜的代碼數(shù)據(jù)傾斜僅發(fā)生在shuffle進程中。以下是可以觸發(fā)shuffle操作的常見運算符：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup和repartition。如果數(shù)據(jù)傾斜，則可能是在代碼中使用這些運算符之一的結(jié)果。Task運行特別慢第一個問題是，數(shù)據(jù)傾斜發(fā)生在第幾個s

4、tage上。如果以Yarn-client模式提交，則可以在本地查看日志，并且日志中當前運行的stage中選擇所需的構(gòu)件。在Yarn-cluster模式下提交時，Spark Web UI可以查看當前是否正在運行到第幾個stage。您還可以在Spark Web UI中詳細查看當前stage中的單個task分配的數(shù)據(jù)量，無論您是在yarn-client模式還是yarn-cluster模式下，都可以進一步了解task分配的數(shù)據(jù)不均勻是否導(dǎo)致了數(shù)據(jù)傾斜。例如，在下圖中，最后第三列顯示了每個task的運行時間。您可能會發(fā)現(xiàn)，某些task的運行速度非?？欤恍鑾酌腌娂纯蛇\行。某些task的運行時間非常慢，運

5、行需要幾分鐘，此時只能在運行時間確定數(shù)據(jù)的斜率。倒計時的第一列還顯示了每個task處理的數(shù)據(jù)的楊怡，運行時間特別短的task只需要處理數(shù)百KB的數(shù)據(jù)，運行時間特別長的task需要處理數(shù)千KB的數(shù)據(jù)，處理的數(shù)據(jù)量減少了10倍。您可以更好地知道此時是否發(fā)生了數(shù)據(jù)傾斜。知道了發(fā)生數(shù)據(jù)傾斜的stage后，必須根據(jù)stage拆分原理計算發(fā)生傾斜的stage的相應(yīng)代碼中具有shuffle類運算符的部分。正確計算需要深入了解火花源代碼的stage和代碼的對應(yīng)關(guān)系，當火花代碼出現(xiàn)一個shuffle類運算符或火花Spark SQL語句出現(xiàn)導(dǎo)致shuffle的語句(例如group by語句)時，您可以根據(jù)該位置確

6、定兩個stage是分開的。這里，我們以火花的基本入門級程序字數(shù)為例，看一下如何用最簡單的方法粗略計算對應(yīng)于Spark的代碼。在下面的示例中，整個代碼中只有一個reduceByKey是引發(fā)shuffle的運算符，因此，您可以將此運算符作為邊界，將前后兩個stage拆分開。Stage0主要在textFile中使用map，并執(zhí)行shuffle write操作。這可以理解為Shuffle write操作，簡單地將pairs RDD上的數(shù)據(jù)分區(qū)，每個task處理的數(shù)據(jù)中的相同密鑰都寫入到同一磁盤文件中。Stage1主要執(zhí)行從reduceByKey到collect的任務(wù)，在stage1中的每個task從頭

7、開始運行時，將首先執(zhí)行shuffle read任務(wù)。執(zhí)行Shuffle read操作的task在stage0的每個task節(jié)點上拖動其處理的關(guān)鍵點，然后對同一關(guān)鍵點執(zhí)行完整集合或join等操作。其中累計key的value值。Stage1運行reduceByKey運算符后，計算最終的wordCounts RDD，運行collect運算符，將所有數(shù)據(jù)拖到Driver中，遍歷輸出并打印輸出。plain view plain copy1.valconf=newparkconf()2.valsc=newparkcontext (conf)3.4.vallines=sc . textfile( HDFS

8、:/. ）5.valwords=lines.flatmap (_ .split(“”)6.val pairs=words . map(_，1)7.valword counts=pairs . reduce bykey(_ _ _)8.9.wordcounts.collect()。foreach (println (_)我想分析字數(shù)程序，告訴您stage分割的最基本原理，以及stage分割后shuffle操作在兩個stage的邊界上是如何執(zhí)行的。然后，您可以了解如何快速找到發(fā)生數(shù)據(jù)傾斜的stage的相應(yīng)代碼部分。例如，如果Spark Web UI或本地log確定stage1中的某些task的執(zhí)行

9、速度特別慢，并且stage1具有數(shù)據(jù)傾斜，則可以通過確認stage1主要包含shuffle類運算符(例如reduceByKey)來確定educeByKey運算符導(dǎo)致的數(shù)據(jù)傾斜問題。例如，如果一個單詞出現(xiàn)了100萬次，另一個單詞出現(xiàn)了10次，則stage1中的一個task處理100萬個數(shù)據(jù)時，整個stage的速度會被此task減慢。任何task以某種方式導(dǎo)致內(nèi)存溢出在這種情況下，查找問題的代碼更容易。建議您直接查看YARN-client模式下本地日志的異常錯誤堆棧，或通過YARN查看yarn-cluster模式下log中的異常錯誤堆棧。通常，異常堆棧信息允許代碼跳轉(zhuǎn)到發(fā)生內(nèi)存溢出的行。然后瀏覽代

10、碼行，通常還會有shuffle類運算符。此操作符很可能導(dǎo)致數(shù)據(jù)傾斜。但是要注意，不能因為不小心泄漏了內(nèi)存就判斷出發(fā)生了數(shù)據(jù)傾斜。因為您編寫的代碼中的錯誤和意外發(fā)生的數(shù)據(jù)異常會導(dǎo)致內(nèi)存溢出。如上所述，您必須查看通過Spark Web UI報告錯誤的stage的單獨task運行時間和分配的數(shù)據(jù)量，以確定是否發(fā)生了內(nèi)存泄漏。查看key中數(shù)據(jù)傾斜的分布知道數(shù)據(jù)傾斜發(fā)生的位置后，通常會執(zhí)行shuffle操作，分析數(shù)據(jù)傾斜的RDD/Hive表以查看關(guān)鍵點分布。這主要是為以后選擇什么技術(shù)方案提供了依據(jù)。徐璐對于不同的關(guān)鍵點分布和徐璐其他“shuffle”操作符組合在一起的不同情況，您可能需要徐璐選擇不同的技

11、術(shù)解決方案。根據(jù)您正在執(zhí)行的任務(wù)，有多種查看關(guān)鍵點分布的方法。1.Spark SQL中的group by，join語句導(dǎo)致數(shù)據(jù)傾斜時，查詢SQL中使用的表的key分布。2.如果在Spark RDD上運行shuffle運算符導(dǎo)致數(shù)據(jù)傾斜，則可以在Spark操作中添加代碼以查看鍵分布，如RDD.countByKey()。然后，對于收集的每個密鑰的出現(xiàn)次數(shù)，可以通過collect/take打印到客戶端來檢查密鑰分布。例如，對于上面提到的字數(shù)程序，如果確定stage1的reduceByKey運算符會導(dǎo)致數(shù)據(jù)傾斜，則需要查看執(zhí)行reduceByKey操作的RDD的鍵分配。在這種情況下，例如pairs R

12、DD。在以下示例中，可以首先對pairs的10%采樣數(shù)據(jù)進行采樣，然后使用countByKey運算符計算每個關(guān)鍵點的發(fā)生次數(shù)，最后在客戶端上遍歷和打印采樣數(shù)據(jù)中單個關(guān)鍵點的發(fā)生次數(shù)。plain view plain copy1.valsampledairs=pairs . sample(false，0.1)2.valsampledordcounts=sampled pairs . count bykey()3.sampledordcounts . foreach(print ln(_)2.數(shù)據(jù)傾斜解決方案解決方案1:使用蜂巢ETL預(yù)處理數(shù)據(jù)案例套用案例：資料傾斜是Hive表格。如果Hive表中

13、的數(shù)據(jù)本身不統(tǒng)一(例如，關(guān)鍵字對應(yīng)于百萬數(shù)據(jù)，其他關(guān)鍵字對應(yīng)于10個數(shù)據(jù))，則此技術(shù)適用于業(yè)務(wù)場景中需要經(jīng)常使用火花對hit表進行分析的情況。方案實施思路：此時，評估是否可以通過Hive預(yù)處理數(shù)據(jù)(即，是否可以通過Hive ETL將數(shù)據(jù)預(yù)聚合到key或預(yù)聯(lián)接到其他表)后，火花操作中要作為目標的數(shù)據(jù)源是預(yù)處理的Hive表，而不是原始Hive表?，F(xiàn)在，Spark作業(yè)不再需要使用原始shuffle類運算符執(zhí)行這些操作，因為已收集數(shù)據(jù)或執(zhí)行了join操作。方案實施原則：由于此方案完全避免在Spark中運行shuffle類運算符，因此從源角度解決了數(shù)據(jù)傾斜。但是要提醒你，這種方式是消除根本。最終，數(shù)據(jù)本

14、身存在不均勻的散布問題，因此在Hive ETL上執(zhí)行組by或join等shuffle操作時，數(shù)據(jù)仍然傾斜，從而降低Hive ETL的速度。為了避免火花程序中的數(shù)據(jù)傾斜，我們只能將數(shù)據(jù)傾斜提前到蜂巢ETL。方案優(yōu)勢：可以簡單方便地實施，從而完全消除數(shù)據(jù)傾斜并顯著提高火花工作的性能。方案缺點：數(shù)據(jù)在hitetl(而不是根本原因)中傾斜。案例實踐經(jīng)驗：在與某些Java系統(tǒng)搭配使用Spark的專案中，如果Java程式碼經(jīng)常呼叫Spark動作，并且Spark動作的效能需求很高，則此案例更為適用。將數(shù)據(jù)提前到上游Hive ETL，每天運行一次更慢，然后每次Java調(diào)用Spark作業(yè)時執(zhí)行速度更快，從而提供

15、更好的用戶體驗。項目實踐經(jīng)驗：此方案主要用于美國集團評論的交互式用戶行為分析系統(tǒng)，該系統(tǒng)允許用戶通過Java web系統(tǒng)提交數(shù)據(jù)分析統(tǒng)計作業(yè)，后端通過Java提交Spark作業(yè)以分析數(shù)據(jù)分析統(tǒng)計。Spark操作必須快，并且必須在可能的10分鐘內(nèi)完成。否則，速度太慢，用戶環(huán)境可能會很差。因此，我們將一些火花工作的shuffle工作提前到了Hive ETL。因此，通過直接使用預(yù)處理spark的Hive臨時表，將spark的shuffle操作降至最低，顯著提高了性能，并將某些操作的性能提高了6倍以上。解決方案2:過濾導(dǎo)致傾斜的少數(shù)關(guān)鍵點案例套用案例：如果導(dǎo)致斜度的索引鍵較少，對計算本身影響不大，則此

16、案例適用。例如，99%的密鑰對應(yīng)于10個數(shù)據(jù)，但只有1個密鑰對應(yīng)于100萬個數(shù)據(jù)，因此數(shù)據(jù)傾斜。方案實施思路：判斷數(shù)據(jù)楊怡，特別是許多關(guān)鍵點，如果任務(wù)執(zhí)行和計算結(jié)果不是特別重要，則簡單地過濾掉其中的幾個關(guān)鍵點。例如，在Spark SQL中，您可以使用where子句過濾這些關(guān)鍵字，或在Spark Core中對RDD運行filter運算符以過濾這些關(guān)鍵字。要在每次作業(yè)運行時過濾之前動態(tài)確定數(shù)據(jù)楊怡中最多的關(guān)鍵點，可以使用sample運算符對RDD進行采樣，然后計算每個關(guān)鍵點的數(shù)量，最后使用數(shù)據(jù)楊怡中最多的關(guān)鍵點進行過濾。方案實施原則：在過濾了數(shù)據(jù)傾斜的鍵后，不能生成數(shù)據(jù)傾斜，因為不參與計算。方案優(yōu)勢：簡單有效，完全避免了數(shù)據(jù)傾斜。場景缺點：適用的場景不多，在大多數(shù)情況下，導(dǎo)致傾斜的關(guān)鍵點仍然很多，并且不僅僅是幾個。情景實踐經(jīng)驗：使用此解決方案解決了項目中的數(shù)據(jù)坡率。一天，Spark作業(yè)運行時突然發(fā)現(xiàn)了OOM，跟蹤發(fā)現(xiàn)蜂巢表的鍵在當天異常地增加了數(shù)據(jù)楊怡爆炸。因此