戰(zhàn)：算子優(yōu)化、數(shù)據(jù)傾斜與集群調(diào)優(yōu))
1. 從一個看起來簡單的WordCount說起但凡接觸過分布式計(jì)算的人幾乎都繞不開WordCount這個案例。它就像編程語言里的Hello World看著簡單但真正動手寫的時候你會發(fā)現(xiàn)里面藏著不少門道。我見過太多人第一次跑WordCount代碼照著示例敲完提交上去結(jié)果要么報錯要么輸出一堆空文件要么結(jié)果跟預(yù)期對不上。問題往往不在邏輯本身而在于對Spark運(yùn)行機(jī)制的理解不夠深。這篇內(nèi)容我打算把WordCount和TopKey這兩個經(jīng)典案例徹底講透。不只是給你一份能跑的Scala代碼而是把每個算子為什么這么選、參數(shù)為什么這么設(shè)、集群上跑的時候會遇到什么坑全部掰開揉碎說清楚。適合已經(jīng)了解Scala基礎(chǔ)語法、想真正把Spark用起來的人。如果你還在糾結(jié)要不要學(xué)Spark那這篇可能不太適合你但如果你已經(jīng)決定要上手并且想少走彎路那接下來的內(nèi)容應(yīng)該能幫你省下不少調(diào)試時間。先明確一下這兩個案例的核心價值。WordCount解決的是詞頻統(tǒng)計(jì)問題本質(zhì)上是把一堆文本拆成單詞然后按單詞分組計(jì)數(shù)。TopKey則是在WordCount的基礎(chǔ)上找出出現(xiàn)頻率最高的前N個詞。這兩個案例覆蓋了Spark核心編程中最常用的算子組合flatMap、map、reduceByKey、sortBy、take等。把它們吃透后面寫更復(fù)雜的業(yè)務(wù)邏輯就是搭積木的事。2. 環(huán)境準(zhǔn)備與項(xiàng)目骨架搭建2.1 依賴版本的選擇邏輯在動手寫代碼之前先把環(huán)境理清楚。Spark的版本迭代很快不同版本之間API有差異尤其是Spark 2.x和3.x之間。我個人的建議是如果是新項(xiàng)目直接上Spark 3.x因?yàn)樗谛阅軆?yōu)化和API簡潔性上都有明顯提升。但如果你所在的環(huán)境還在用Spark 2.x那也沒關(guān)系WordCount和TopKey這兩個案例的核心API在兩個版本里基本一致。Scala版本方面Spark 3.x默認(rèn)用的是Scala 2.12Spark 2.4之前用的是2.11。這個必須對齊否則編譯的時候會報NoSuchMethodError之類的錯誤而且這種錯誤往往在運(yùn)行時才暴露排查起來很煩。我踩過這個坑當(dāng)時本地Scala是2.13集群上是2.12代碼本地跑得好好的一提交就掛查了半天才發(fā)現(xiàn)是版本不匹配。構(gòu)建工具用Maven或者sbt都行。sbt對Scala項(xiàng)目更友好依賴管理更自然Maven的話配置稍微繁瑣一點(diǎn)但團(tuán)隊(duì)協(xié)作時更通用。下面以Maven為例給出核心依賴配置dependencies dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.3.0/version /dependency dependency groupIdorg.scala-lang/groupId artifactIdscala-library/artifactId version2.12.15/version /dependency /dependencies注意artifactId里的_2.12后綴這個數(shù)字必須和你的Scala版本一致。很多人復(fù)制依賴的時候只看版本號忽略了后綴結(jié)果編譯報錯找不到類。2.2 項(xiàng)目目錄結(jié)構(gòu)目錄結(jié)構(gòu)不用太復(fù)雜但建議按職責(zé)分一下。我習(xí)慣這樣組織src/main/scala/ com/example/spark/ WordCount.scala TopKey.scala util/ SparkSessionBuilder.scala把創(chuàng)建SparkSession的邏輯單獨(dú)抽出來是因?yàn)閷?shí)際項(xiàng)目中幾乎每個任務(wù)都需要初始化SparkSession而且配置參數(shù)往往有共性。抽成一個工具類后面寫新任務(wù)的時候直接調(diào)用省得重復(fù)寫一堆配置。package com.example.spark.util import org.apache.spark.sql.SparkSession object SparkSessionBuilder { def create(appName: String, master: String local[*]): SparkSession { SparkSession.builder() .appName(appName) .master(master) .config(spark.sql.shuffle.partitions, 10) .getOrCreate() } }這里我特意設(shè)了spark.sql.shuffle.partitions為10。默認(rèn)值是200對于小數(shù)據(jù)集來說200個分區(qū)意味著200個任務(wù)每個任務(wù)處理的數(shù)據(jù)量極小調(diào)度開銷反而成了瓶頸。本地測試的時候設(shè)小一點(diǎn)能明顯感覺到速度提升。當(dāng)然生產(chǎn)環(huán)境要根據(jù)數(shù)據(jù)量調(diào)整這個后面會細(xì)說。3. WordCount的Scala實(shí)現(xiàn)與算子拆解3.1 完整代碼與逐行解讀先上完整代碼然后逐段拆解package com.example.spark import com.example.spark.util.SparkSessionBuilder object WordCount { def main(args: Array[String]): Unit { val spark SparkSessionBuilder.create(WordCount) val sc spark.sparkContext val inputPath args(0) val outputPath args(1) val textRDD sc.textFile(inputPath) val wordCounts textRDD .flatMap(line line.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) wordCounts.saveAsTextFile(outputPath) spark.stop() } }這段代碼看著簡單但每一行都有講究。sc.textFile(inputPath)讀取文件返回的是一個RDD[String]每個元素是文件中的一行。這里有個細(xì)節(jié)如果輸入路徑是本地文件系統(tǒng)需要確保集群上每個節(jié)點(diǎn)都能訪問到如果是HDFS路徑那就沒這個問題。我見過有人在本地用file:///路徑測試提交到集群后報FileNotFoundException就是因?yàn)槁窂經(jīng)]改成HDFS。flatMap(line line.split(\\s))把每行拆成單詞。用\\s而不是簡單的空格是因?yàn)閷?shí)際文本里可能有多個連續(xù)空格、制表符、換行符等。flatMap和map的區(qū)別在于flatMap會把拆分后的數(shù)組拍平返回一個包含所有單詞的RDD而不是RDD[Array[String]]。filter(_.nonEmpty)過濾掉空字符串。這個步驟容易被忽略但如果文本里有連續(xù)空格split之后會產(chǎn)生空字符串不過濾的話最后統(tǒng)計(jì)結(jié)果里會出現(xiàn)一個空key看著很別扭。map(word (word, 1))把每個單詞映射成鍵值對這是為后面的reduceByKey做準(zhǔn)備。Spark的reduceByKey要求輸入是(key, value)形式的RDD。reduceByKey(_ _)按key分組對value做累加。這里用reduceByKey而不是groupByKey是因?yàn)閞educeByKey會在map端做本地聚合減少shuffle的數(shù)據(jù)量。對于詞頻統(tǒng)計(jì)這種場景reduceByKey的性能優(yōu)勢非常明顯。3.2 reduceByKey和groupByKey的性能差異很多人寫WordCount的時候習(xí)慣用groupByKey然后mapValues(_.sum)邏輯上沒問題但性能差很多。我做過一個測試同樣的數(shù)據(jù)集reduceByKey比groupByKey快了將近三倍。原因在于shuffle機(jī)制。groupByKey會把所有相同key的value拉到同一個分區(qū)不做任何預(yù)聚合。如果某個key對應(yīng)的數(shù)據(jù)量特別大還會導(dǎo)致數(shù)據(jù)傾斜某個task處理的數(shù)據(jù)量遠(yuǎn)超其他task拖慢整個作業(yè)。reduceByKey則會在map端先做一次本地聚合相當(dāng)于把局部求和提前做了shuffle的時候只傳輸聚合后的結(jié)果數(shù)據(jù)量小很多。注意如果業(yè)務(wù)邏輯不是簡單的求和而是需要保留所有原始value比如要計(jì)算中位數(shù)那groupByKey還是得用。但純粹做計(jì)數(shù)、求和這類聚合操作優(yōu)先選reduceByKey。3.3 輸出結(jié)果的處理與常見問題saveAsTextFile(outputPath)把結(jié)果寫到指定路徑。這里有幾個坑第一輸出路徑不能已經(jīng)存在。Spark會檢查路徑如果目錄已存在直接拋異常。調(diào)試的時候反復(fù)運(yùn)行每次都要手動刪輸出目錄很煩?？梢栽诖a里加一段判斷或者用hadoop fs -rm -r先清理。第二輸出是一個目錄里面包含多個part文件數(shù)量等于最終RDD的分區(qū)數(shù)。如果分區(qū)數(shù)是200就會生成200個文件哪怕數(shù)據(jù)量很小。這時候可以用coalesce(1)把結(jié)果合并到一個文件里但要注意coalesce(1)會把所有數(shù)據(jù)拉到一個節(jié)點(diǎn)數(shù)據(jù)量大時可能OOM。更穩(wěn)妥的做法是repartition(1)但同樣有性能問題。我的建議是小結(jié)果集用coalesce(1)大結(jié)果集保持多文件輸出后續(xù)用其他工具合并。第三輸出格式是(word, count)的字符串形式因?yàn)閟aveAsTextFile調(diào)用的是元素的toString方法。如果后續(xù)要程序化讀取這種格式不方便解析?？梢钥紤]用saveAsObjectFile或者轉(zhuǎn)成DataFrame后寫CSV/Parquet。4. TopKey的實(shí)現(xiàn)思路與排序策略4.1 從WordCount到TopKey的演進(jìn)TopKey的需求是在詞頻統(tǒng)計(jì)的基礎(chǔ)上找出出現(xiàn)次數(shù)最多的前N個詞。這個需求在實(shí)際場景中很常見比如分析熱搜詞、統(tǒng)計(jì)高頻錯誤碼、找出訪問量最大的URL等。實(shí)現(xiàn)思路有兩種。第一種是先做完整的WordCount然后對結(jié)果按count降序排列取前N個。第二種是用takeOrdered或者top算子直接在RDD層面做。兩種方式各有適用場景下面分別說。4.2 基于sortBy的全局排序方案先看第一種也是最直觀的方案package com.example.spark import com.example.spark.util.SparkSessionBuilder object TopKey { def main(args: Array[String]): Unit { val spark SparkSessionBuilder.create(TopKey) val sc spark.sparkContext val inputPath args(0) val topN args(1).toInt val textRDD sc.textFile(inputPath) val topKeys textRDD .flatMap(_.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) .sortBy(_._2, ascending false) .take(topN) topKeys.foreach(println) spark.stop() } }這段代碼的核心是sortBy(_._2, ascending false)。sortBy會對整個RDD進(jìn)行全局排序底層通過rangePartitioner實(shí)現(xiàn)。排序完成后take(topN)取前N個元素。這里有個性能問題sortBy是全局排序會觸發(fā)一次全量shuffle把所有數(shù)據(jù)按照count重新分區(qū)。如果數(shù)據(jù)量很大這一步開銷不小。而且我們只需要前N個卻對所有數(shù)據(jù)做了排序有點(diǎn)浪費(fèi)。4.3 takeOrdered與top的取舍Spark提供了takeOrdered和top兩個算子專門用于取前N個元素不需要全局排序。它們的區(qū)別在于top(n)取最大的N個元素使用默認(rèn)的降序排列。takeOrdered(n)取最小的N個元素使用默認(rèn)的升序排列。對于TopKey場景我們要取count最大的N個詞可以用top但top比較的是整個元素而我們的元素是(word, count)這樣的元組。元組的默認(rèn)排序是先比第一個元素再比第二個。我們想按count排序就需要提供一個自定義的Ordering。val topKeys textRDD .flatMap(_.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) .top(topN)(Ordering.by(_._2))Ordering.by(_._2)指定按照元組的第二個元素也就是count來比較。top內(nèi)部會先在每個分區(qū)取前N個然后匯總效率比全局排序高很多。但top有一個限制它返回的是一個數(shù)組數(shù)據(jù)量大的時候這個數(shù)組會全部拉到Driver端。如果N很大Driver端可能內(nèi)存不夠。所以top適合N比較小的場景比如取前10、前100。如果N是幾千甚至幾萬還是用sortBy然后take更穩(wěn)妥或者用sortBy后limit再寫文件。4.4 處理數(shù)據(jù)傾斜的實(shí)戰(zhàn)技巧TopKey場景下數(shù)據(jù)傾斜是一個繞不開的問題。如果某個詞出現(xiàn)頻率極高比如的、是這類停用詞那么reduceByKey的時候這個key對應(yīng)的數(shù)據(jù)量會遠(yuǎn)超其他key導(dǎo)致某個task成為瓶頸。我遇到過一個案例日志分析任務(wù)里某個錯誤碼占了總?cè)罩玖康?0%reduce階段直接卡住。解決辦法有幾個第一預(yù)處理階段過濾掉明顯的停用詞。如果是文本分析維護(hù)一個停用詞表在flatMap之后加一個filter把停用詞去掉。這個最簡單效果也最直接。第二加鹽打散。給key加上隨機(jī)前綴比如(word _ Random.nextInt(10), 1)先做一次聚合然后再去掉前綴做第二次聚合。這樣能把一個熱key拆成10個分散到不同task上。代價是多了一次shuffle但能解決傾斜問題。第三提高并行度。增加shuffle的分區(qū)數(shù)讓每個task處理的數(shù)據(jù)量小一點(diǎn)?？梢酝ㄟ^spark.sql.shuffle.partitions或者reduceByKey(_ _, numPartitions)來指定。提示加鹽方案會增加代碼復(fù)雜度如果不是傾斜特別嚴(yán)重優(yōu)先考慮過濾停用詞和調(diào)整并行度。加鹽適合傾斜非常極端、且無法通過過濾解決的場景。5. 集群提交與參數(shù)調(diào)優(yōu)的實(shí)操細(xì)節(jié)5.1 spark-submit的常用參數(shù)代碼寫好了本地跑通只是第一步真正要上集群得用spark-submit提交。下面是一個典型的提交命令spark-submit \ --class com.example.spark.WordCount \ --master yarn \ --deploy-mode cluster \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 10 \ --driver-memory 2g \ wordcount.jar \ hdfs:///input/text.txt \ hdfs:///output/wordcount每個參數(shù)都有講究。--executor-memory決定每個Executor的堆內(nèi)存--executor-cores決定每個Executor能并行跑幾個task。--num-executors是Executor的數(shù)量。這三個參數(shù)共同決定了作業(yè)的總并行度num-executors * executor-cores。很多人調(diào)優(yōu)的時候只盯著內(nèi)存加忽略了cores。實(shí)際上如果cores太少每個Executor只能串行跑task內(nèi)存再大也用不上。一般來說每個Executor配2到5個cores比較合適太少浪費(fèi)內(nèi)存太多會導(dǎo)致GC壓力大。--driver-memory容易被忽略。如果代碼里有collect、take這類把數(shù)據(jù)拉到Driver端的操作Driver內(nèi)存不夠會直接OOM。TopKey場景下如果用top取前N個N很大時就要注意Driver內(nèi)存。5.2 內(nèi)存配置的計(jì)算邏輯Executor內(nèi)存怎么定一個粗略的估算方法是總數(shù)據(jù)量除以總并行度再乘以一個安全系數(shù)。假設(shè)輸入數(shù)據(jù)是10GBnum-executors10executor-cores2那么總并行度是20。每個task平均處理500MB數(shù)據(jù)?？紤]到shuffle、緩存、對象開銷每個task至少需要1GB內(nèi)存那么每個Executor跑2個task需要2GB用于計(jì)算再加上堆外內(nèi)存和系統(tǒng)預(yù)留executor-memory設(shè)4GB比較穩(wěn)妥。當(dāng)然這只是估算實(shí)際還要看數(shù)據(jù)特征。如果數(shù)據(jù)里字符串特別多對象開銷會更大如果有大量shuffle內(nèi)存需求也會上升。我的經(jīng)驗(yàn)是先按估算值跑一次看Spark UI里的GC時間和spill情況再針對性調(diào)整。5.3 提交后常見報錯與排查提交到集群后最常見的報錯有這么幾類第一類ClassNotFoundException。通常是依賴沒打進(jìn)去。用Maven的shade插件打一個fat jar把所有依賴都包含進(jìn)去。注意Spark本身的依賴要設(shè)成provided因?yàn)榧荷弦呀?jīng)有了打進(jìn)去反而可能沖突。第二類Container killed by YARN for exceeding memory limits。這是Executor內(nèi)存超了?？赡苁菙?shù)據(jù)傾斜導(dǎo)致某個task處理的數(shù)據(jù)量特別大也可能是內(nèi)存配置本身就不夠。先看Spark UI里各task的處理數(shù)據(jù)量確認(rèn)是不是傾斜再決定是加內(nèi)存還是解決傾斜。第三類Shuffle fetch failed。通常是Executor掛了導(dǎo)致shuffle數(shù)據(jù)丟失。可以開啟spark.shuffle.service.enabled讓shuffle數(shù)據(jù)由外部服務(wù)管理Executor掛了也不影響。另外spark.network.timeout可以適當(dāng)調(diào)大避免網(wǎng)絡(luò)抖動導(dǎo)致的fetch失敗。第四類輸出目錄已存在。前面提過Spark不允許輸出到已存在的目錄?？梢栽诖a里加判斷或者每次提交前手動清理。6. 從案例到生產(chǎn)的思維轉(zhuǎn)變6.1 本地模式與集群模式的差異本地模式跑通不代表集群上沒問題。我總結(jié)了幾點(diǎn)差異都是實(shí)際踩過的第一文件路徑。本地用file:///集群上要用HDFS路徑。如果代碼里寫死了本地路徑提交上去必然報錯。建議把路徑作為參數(shù)傳入不要硬編碼。第二序列化。本地模式下很多對象不需要序列化就能傳遞集群模式下所有在Executor上執(zhí)行的對象都必須可序列化。如果代碼里用了自定義類記得實(shí)現(xiàn)Serializable接口。否則會報NotSerializableException而且報錯信息往往指向一個看起來無關(guān)的地方排查起來很費(fèi)勁。第三日志級別。本地跑的時候控制臺輸出很清晰集群上跑日志分散在各個Executor的container里需要去YARN的日志聚合里看。建議在代碼里把日志級別調(diào)成WARN減少不必要的輸出方便定位問題。6.2 代碼健壯性的幾個習(xí)慣寫Spark代碼有幾個習(xí)慣能幫你省很多事習(xí)慣一參數(shù)校驗(yàn)。輸入路徑、輸出路徑、TopN這些參數(shù)在main函數(shù)開頭就校驗(yàn)不合法直接退出別等到運(yùn)行到一半才報錯。習(xí)慣二資源關(guān)閉。spark.stop()要確保執(zhí)行最好放在finally塊里。雖然SparkSession不關(guān)也不會導(dǎo)致嚴(yán)重問題但長期運(yùn)行的服務(wù)里不關(guān)會占著資源。習(xí)慣三異常捕獲。對于可能失敗的操作比如讀寫HDFS加try-catch把異常信息打出來方便排查。但不要吞異常捕獲了要處理或者重新拋出。習(xí)慣四配置外置。把路徑、并行度、內(nèi)存這些配置抽到配置文件或者命令行參數(shù)里不要硬編碼。這樣換環(huán)境的時候不用改代碼重新提交就行。6.3 性能監(jiān)控與調(diào)優(yōu)的切入點(diǎn)作業(yè)跑起來之后怎么知道性能好不好看Spark UI。重點(diǎn)看幾個指標(biāo)Task的duration分布如果大部分task很快少數(shù)task特別慢說明有數(shù)據(jù)傾斜。GC時間如果GC時間占比超過10%說明內(nèi)存不夠需要加內(nèi)存或者減少緩存。Shuffle read/write如果shuffle數(shù)據(jù)量特別大考慮優(yōu)化算子減少shuffle。Spill如果出現(xiàn)spill說明內(nèi)存不足以容納當(dāng)前分區(qū)的數(shù)據(jù)需要加內(nèi)存或者增加分區(qū)數(shù)。調(diào)優(yōu)是一個迭代的過程沒有一勞永逸的配置。我的做法是先保證功能正確再根據(jù)監(jiān)控數(shù)據(jù)逐步優(yōu)化。每次只調(diào)一個參數(shù)觀察效果避免同時改多個參數(shù)導(dǎo)致無法定位問題。7. 一些容易被忽略的細(xì)節(jié)7.1 文本編碼問題讀取文本文件時默認(rèn)是UTF-8編碼。如果文件是GBK編碼中文會出現(xiàn)亂碼。解決辦法是在讀取時指定編碼或者先用工具轉(zhuǎn)成UTF-8。sc.textFile本身不提供編碼參數(shù)可以用sparkContext.hadoopConfiguration設(shè)置或者用wholeTextFiles讀取后手動解碼。7.2 空文件與空分區(qū)的處理如果輸入目錄里有空文件textFile會正常讀取但對應(yīng)的分區(qū)里沒有數(shù)據(jù)。這本身不是問題但如果后續(xù)有mapPartitions之類的操作要處理空迭代器的情況否則可能報錯。7.3 結(jié)果驗(yàn)證的方法跑完WordCount怎么驗(yàn)證結(jié)果對不對我的做法是先用一個小樣本數(shù)據(jù)手動算一遍預(yù)期結(jié)果然后跟Spark的輸出對比。確認(rèn)邏輯沒問題后再上全量數(shù)據(jù)。全量數(shù)據(jù)跑完后可以抽樣檢查幾個key的count跟原始數(shù)據(jù)里的出現(xiàn)次數(shù)對一下。對于TopKey驗(yàn)證方法是先跑完整的WordCount把結(jié)果按count排序取前N個跟TopKey的輸出對比。如果一致說明TopKey的邏輯沒問題。7.4 關(guān)于Scala的隱式轉(zhuǎn)換Scala的隱式轉(zhuǎn)換在Spark代碼里用得很多比如reduceByKey需要(K, V)類型的RDD而map返回的是普通RDD中間就涉及隱式轉(zhuǎn)換。如果import不對編譯會報錯。常見的import包括org.apache.spark.SparkContext._它包含了這些隱式轉(zhuǎn)換。Spark 2.x之后很多隱式轉(zhuǎn)換移到了org.apache.spark.rdd.RDD的伴生對象里import方式有變化需要注意版本差異。8. 從這兩個案例延伸出去WordCount和TopKey雖然簡單但它們覆蓋的算子組合可以延伸到很多場景。比如把單詞換成URL就是訪問量統(tǒng)計(jì)把單詞換成錯誤碼就是錯誤分布分析把count換成金額就是銷售額聚合。核心模式是一樣的拆分、映射、聚合、排序。掌握了這個模式再去看Spark的其他算子比如aggregateByKey、combineByKey、foldByKey會發(fā)現(xiàn)它們只是聚合方式不同底層邏輯是相通的。reduceByKey要求聚合前后的類型一致aggregateByKey允許聚合前后類型不同combineByKey則更靈活可以指定初始創(chuàng)建、分區(qū)內(nèi)合并、分區(qū)間合并三個函數(shù)。我在實(shí)際項(xiàng)目中用得最多的是reduceByKey和aggregateByKey。前者適合簡單的求和、計(jì)數(shù)后者適合求平均值、求TopN這類需要中間狀態(tài)的聚合。比如求每個用戶的平均消費(fèi)金額就可以用aggregateByKey初始值是(0.0, 0)分區(qū)內(nèi)合并是累加金額和次數(shù)分區(qū)間合并也是累加最后再相除。最后分享一個小技巧寫Spark代碼的時候盡量用reduceByKey替代groupByKey用mapPartitions替代map當(dāng)需要在每個分區(qū)內(nèi)做初始化操作時用filter盡早過濾掉不需要的數(shù)據(jù)。這些習(xí)慣能顯著提升作業(yè)性能尤其是在數(shù)據(jù)量大的時候。 Spark WordCount與TopKey實(shí)戰(zhàn)：算子優(yōu)化、數(shù)據(jù)傾斜與集群調(diào)優(yōu)
1. 從一個看起來簡單的WordCount說起但凡接觸過分布式計(jì)算的人幾乎都繞不開WordCount這個案例。它就像編程語言里的Hello World看著簡單但真正動手寫的時候你會發(fā)現(xiàn)里面藏著不少門道。我見過太多人第一次跑WordCount代碼照著示例敲完提交上去結(jié)果要么報錯要么輸出一堆空文件要么結(jié)果跟預(yù)期對不上。問題往往不在邏輯本身而在于對Spark運(yùn)行機(jī)制的理解不夠深。這篇內(nèi)容我打算把WordCount和TopKey這兩個經(jīng)典案例徹底講透。不只是給你一份能跑的Scala代碼而是把每個算子為什么這么選、參數(shù)為什么這么設(shè)、集群上跑的時候會遇到什么坑全部掰開揉碎說清楚。適合已經(jīng)了解Scala基礎(chǔ)語法、想真正把Spark用起來的人。如果你還在糾結(jié)要不要學(xué)Spark那這篇可能不太適合你但如果你已經(jīng)決定要上手并且想少走彎路那接下來的內(nèi)容應(yīng)該能幫你省下不少調(diào)試時間。先明確一下這兩個案例的核心價值。WordCount解決的是詞頻統(tǒng)計(jì)問題本質(zhì)上是把一堆文本拆成單詞然后按單詞分組計(jì)數(shù)。TopKey則是在WordCount的基礎(chǔ)上找出出現(xiàn)頻率最高的前N個詞。這兩個案例覆蓋了Spark核心編程中最常用的算子組合flatMap、map、reduceByKey、sortBy、take等。把它們吃透后面寫更復(fù)雜的業(yè)務(wù)邏輯就是搭積木的事。2. 環(huán)境準(zhǔn)備與項(xiàng)目骨架搭建2.1 依賴版本的選擇邏輯在動手寫代碼之前先把環(huán)境理清楚。Spark的版本迭代很快不同版本之間API有差異尤其是Spark 2.x和3.x之間。我個人的建議是如果是新項(xiàng)目直接上Spark 3.x因?yàn)樗谛阅軆?yōu)化和API簡潔性上都有明顯提升。但如果你所在的環(huán)境還在用Spark 2.x那也沒關(guān)系WordCount和TopKey這兩個案例的核心API在兩個版本里基本一致。Scala版本方面Spark 3.x默認(rèn)用的是Scala 2.12Spark 2.4之前用的是2.11。這個必須對齊否則編譯的時候會報NoSuchMethodError之類的錯誤而且這種錯誤往往在運(yùn)行時才暴露排查起來很煩。我踩過這個坑當(dāng)時本地Scala是2.13集群上是2.12代碼本地跑得好好的一提交就掛查了半天才發(fā)現(xiàn)是版本不匹配。構(gòu)建工具用Maven或者sbt都行。sbt對Scala項(xiàng)目更友好依賴管理更自然Maven的話配置稍微繁瑣一點(diǎn)但團(tuán)隊(duì)協(xié)作時更通用。下面以Maven為例給出核心依賴配置dependencies dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.3.0/version /dependency dependency groupIdorg.scala-lang/groupId artifactIdscala-library/artifactId version2.12.15/version /dependency /dependencies注意artifactId里的_2.12后綴這個數(shù)字必須和你的Scala版本一致。很多人復(fù)制依賴的時候只看版本號忽略了后綴結(jié)果編譯報錯找不到類。2.2 項(xiàng)目目錄結(jié)構(gòu)目錄結(jié)構(gòu)不用太復(fù)雜但建議按職責(zé)分一下。我習(xí)慣這樣組織src/main/scala/ com/example/spark/ WordCount.scala TopKey.scala util/ SparkSessionBuilder.scala把創(chuàng)建SparkSession的邏輯單獨(dú)抽出來是因?yàn)閷?shí)際項(xiàng)目中幾乎每個任務(wù)都需要初始化SparkSession而且配置參數(shù)往往有共性。抽成一個工具類后面寫新任務(wù)的時候直接調(diào)用省得重復(fù)寫一堆配置。package com.example.spark.util import org.apache.spark.sql.SparkSession object SparkSessionBuilder { def create(appName: String, master: String local[*]): SparkSession { SparkSession.builder() .appName(appName) .master(master) .config(spark.sql.shuffle.partitions, 10) .getOrCreate() } }這里我特意設(shè)了spark.sql.shuffle.partitions為10。默認(rèn)值是200對于小數(shù)據(jù)集來說200個分區(qū)意味著200個任務(wù)每個任務(wù)處理的數(shù)據(jù)量極小調(diào)度開銷反而成了瓶頸。本地測試的時候設(shè)小一點(diǎn)能明顯感覺到速度提升。當(dāng)然生產(chǎn)環(huán)境要根據(jù)數(shù)據(jù)量調(diào)整這個后面會細(xì)說。3. WordCount的Scala實(shí)現(xiàn)與算子拆解3.1 完整代碼與逐行解讀先上完整代碼然后逐段拆解package com.example.spark import com.example.spark.util.SparkSessionBuilder object WordCount { def main(args: Array[String]): Unit { val spark SparkSessionBuilder.create(WordCount) val sc spark.sparkContext val inputPath args(0) val outputPath args(1) val textRDD sc.textFile(inputPath) val wordCounts textRDD .flatMap(line line.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) wordCounts.saveAsTextFile(outputPath) spark.stop() } }這段代碼看著簡單但每一行都有講究。sc.textFile(inputPath)讀取文件返回的是一個RDD[String]每個元素是文件中的一行。這里有個細(xì)節(jié)如果輸入路徑是本地文件系統(tǒng)需要確保集群上每個節(jié)點(diǎn)都能訪問到如果是HDFS路徑那就沒這個問題。我見過有人在本地用file:///路徑測試提交到集群后報FileNotFoundException就是因?yàn)槁窂經(jīng)]改成HDFS。flatMap(line line.split(\\s))把每行拆成單詞。用\\s而不是簡單的空格是因?yàn)閷?shí)際文本里可能有多個連續(xù)空格、制表符、換行符等。flatMap和map的區(qū)別在于flatMap會把拆分后的數(shù)組拍平返回一個包含所有單詞的RDD而不是RDD[Array[String]]。filter(_.nonEmpty)過濾掉空字符串。這個步驟容易被忽略但如果文本里有連續(xù)空格split之后會產(chǎn)生空字符串不過濾的話最后統(tǒng)計(jì)結(jié)果里會出現(xiàn)一個空key看著很別扭。map(word (word, 1))把每個單詞映射成鍵值對這是為后面的reduceByKey做準(zhǔn)備。Spark的reduceByKey要求輸入是(key, value)形式的RDD。reduceByKey(_ _)按key分組對value做累加。這里用reduceByKey而不是groupByKey是因?yàn)閞educeByKey會在map端做本地聚合減少shuffle的數(shù)據(jù)量。對于詞頻統(tǒng)計(jì)這種場景reduceByKey的性能優(yōu)勢非常明顯。3.2 reduceByKey和groupByKey的性能差異很多人寫WordCount的時候習(xí)慣用groupByKey然后mapValues(_.sum)邏輯上沒問題但性能差很多。我做過一個測試同樣的數(shù)據(jù)集reduceByKey比groupByKey快了將近三倍。原因在于shuffle機(jī)制。groupByKey會把所有相同key的value拉到同一個分區(qū)不做任何預(yù)聚合。如果某個key對應(yīng)的數(shù)據(jù)量特別大還會導(dǎo)致數(shù)據(jù)傾斜某個task處理的數(shù)據(jù)量遠(yuǎn)超其他task拖慢整個作業(yè)。reduceByKey則會在map端先做一次本地聚合相當(dāng)于把局部求和提前做了shuffle的時候只傳輸聚合后的結(jié)果數(shù)據(jù)量小很多。注意如果業(yè)務(wù)邏輯不是簡單的求和而是需要保留所有原始value比如要計(jì)算中位數(shù)那groupByKey還是得用。但純粹做計(jì)數(shù)、求和這類聚合操作優(yōu)先選reduceByKey。3.3 輸出結(jié)果的處理與常見問題saveAsTextFile(outputPath)把結(jié)果寫到指定路徑。這里有幾個坑第一輸出路徑不能已經(jīng)存在。Spark會檢查路徑如果目錄已存在直接拋異常。調(diào)試的時候反復(fù)運(yùn)行每次都要手動刪輸出目錄很煩?？梢栽诖a里加一段判斷或者用hadoop fs -rm -r先清理。第二輸出是一個目錄里面包含多個part文件數(shù)量等于最終RDD的分區(qū)數(shù)。如果分區(qū)數(shù)是200就會生成200個文件哪怕數(shù)據(jù)量很小。這時候可以用coalesce(1)把結(jié)果合并到一個文件里但要注意coalesce(1)會把所有數(shù)據(jù)拉到一個節(jié)點(diǎn)數(shù)據(jù)量大時可能OOM。更穩(wěn)妥的做法是repartition(1)但同樣有性能問題。我的建議是小結(jié)果集用coalesce(1)大結(jié)果集保持多文件輸出后續(xù)用其他工具合并。第三輸出格式是(word, count)的字符串形式因?yàn)閟aveAsTextFile調(diào)用的是元素的toString方法。如果后續(xù)要程序化讀取這種格式不方便解析?？梢钥紤]用saveAsObjectFile或者轉(zhuǎn)成DataFrame后寫CSV/Parquet。4. TopKey的實(shí)現(xiàn)思路與排序策略4.1 從WordCount到TopKey的演進(jìn)TopKey的需求是在詞頻統(tǒng)計(jì)的基礎(chǔ)上找出出現(xiàn)次數(shù)最多的前N個詞。這個需求在實(shí)際場景中很常見比如分析熱搜詞、統(tǒng)計(jì)高頻錯誤碼、找出訪問量最大的URL等。實(shí)現(xiàn)思路有兩種。第一種是先做完整的WordCount然后對結(jié)果按count降序排列取前N個。第二種是用takeOrdered或者top算子直接在RDD層面做。兩種方式各有適用場景下面分別說。4.2 基于sortBy的全局排序方案先看第一種也是最直觀的方案package com.example.spark import com.example.spark.util.SparkSessionBuilder object TopKey { def main(args: Array[String]): Unit { val spark SparkSessionBuilder.create(TopKey) val sc spark.sparkContext val inputPath args(0) val topN args(1).toInt val textRDD sc.textFile(inputPath) val topKeys textRDD .flatMap(_.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) .sortBy(_._2, ascending false) .take(topN) topKeys.foreach(println) spark.stop() } }這段代碼的核心是sortBy(_._2, ascending false)。sortBy會對整個RDD進(jìn)行全局排序底層通過rangePartitioner實(shí)現(xiàn)。排序完成后take(topN)取前N個元素。這里有個性能問題sortBy是全局排序會觸發(fā)一次全量shuffle把所有數(shù)據(jù)按照count重新分區(qū)。如果數(shù)據(jù)量很大這一步開銷不小。而且我們只需要前N個卻對所有數(shù)據(jù)做了排序有點(diǎn)浪費(fèi)。4.3 takeOrdered與top的取舍Spark提供了takeOrdered和top兩個算子專門用于取前N個元素不需要全局排序。它們的區(qū)別在于top(n)取最大的N個元素使用默認(rèn)的降序排列。takeOrdered(n)取最小的N個元素使用默認(rèn)的升序排列。對于TopKey場景我們要取count最大的N個詞可以用top但top比較的是整個元素而我們的元素是(word, count)這樣的元組。元組的默認(rèn)排序是先比第一個元素再比第二個。我們想按count排序就需要提供一個自定義的Ordering。val topKeys textRDD .flatMap(_.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) .top(topN)(Ordering.by(_._2))Ordering.by(_._2)指定按照元組的第二個元素也就是count來比較。top內(nèi)部會先在每個分區(qū)取前N個然后匯總效率比全局排序高很多。但top有一個限制它返回的是一個數(shù)組數(shù)據(jù)量大的時候這個數(shù)組會全部拉到Driver端。如果N很大Driver端可能內(nèi)存不夠。所以top適合N比較小的場景比如取前10、前100。如果N是幾千甚至幾萬還是用sortBy然后take更穩(wěn)妥或者用sortBy后limit再寫文件。4.4 處理數(shù)據(jù)傾斜的實(shí)戰(zhàn)技巧TopKey場景下數(shù)據(jù)傾斜是一個繞不開的問題。如果某個詞出現(xiàn)頻率極高比如的、是這類停用詞那么reduceByKey的時候這個key對應(yīng)的數(shù)據(jù)量會遠(yuǎn)超其他key導(dǎo)致某個task成為瓶頸。我遇到過一個案例日志分析任務(wù)里某個錯誤碼占了總?cè)罩玖康?0%reduce階段直接卡住。解決辦法有幾個第一預(yù)處理階段過濾掉明顯的停用詞。如果是文本分析維護(hù)一個停用詞表在flatMap之后加一個filter把停用詞去掉。這個最簡單效果也最直接。第二加鹽打散。給key加上隨機(jī)前綴比如(word _ Random.nextInt(10), 1)先做一次聚合然后再去掉前綴做第二次聚合。這樣能把一個熱key拆成10個分散到不同task上。代價是多了一次shuffle但能解決傾斜問題。第三提高并行度。增加shuffle的分區(qū)數(shù)讓每個task處理的數(shù)據(jù)量小一點(diǎn)?？梢酝ㄟ^spark.sql.shuffle.partitions或者reduceByKey(_ _, numPartitions)來指定。提示加鹽方案會增加代碼復(fù)雜度如果不是傾斜特別嚴(yán)重優(yōu)先考慮過濾停用詞和調(diào)整并行度。加鹽適合傾斜非常極端、且無法通過過濾解決的場景。5. 集群提交與參數(shù)調(diào)優(yōu)的實(shí)操細(xì)節(jié)5.1 spark-submit的常用參數(shù)代碼寫好了本地跑通只是第一步真正要上集群得用spark-submit提交。下面是一個典型的提交命令spark-submit \ --class com.example.spark.WordCount \ --master yarn \ --deploy-mode cluster \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 10 \ --driver-memory 2g \ wordcount.jar \ hdfs:///input/text.txt \ hdfs:///output/wordcount每個參數(shù)都有講究。--executor-memory決定每個Executor的堆內(nèi)存--executor-cores決定每個Executor能并行跑幾個task。--num-executors是Executor的數(shù)量。這三個參數(shù)共同決定了作業(yè)的總并行度num-executors * executor-cores。很多人調(diào)優(yōu)的時候只盯著內(nèi)存加忽略了cores。實(shí)際上如果cores太少每個Executor只能串行跑task內(nèi)存再大也用不上。一般來說每個Executor配2到5個cores比較合適太少浪費(fèi)內(nèi)存太多會導(dǎo)致GC壓力大。--driver-memory容易被忽略。如果代碼里有collect、take這類把數(shù)據(jù)拉到Driver端的操作Driver內(nèi)存不夠會直接OOM。TopKey場景下如果用top取前N個N很大時就要注意Driver內(nèi)存。5.2 內(nèi)存配置的計(jì)算邏輯Executor內(nèi)存怎么定一個粗略的估算方法是總數(shù)據(jù)量除以總并行度再乘以一個安全系數(shù)。假設(shè)輸入數(shù)據(jù)是10GBnum-executors10executor-cores2那么總并行度是20。每個task平均處理500MB數(shù)據(jù)?？紤]到shuffle、緩存、對象開銷每個task至少需要1GB內(nèi)存那么每個Executor跑2個task需要2GB用于計(jì)算再加上堆外內(nèi)存和系統(tǒng)預(yù)留executor-memory設(shè)4GB比較穩(wěn)妥。當(dāng)然這只是估算實(shí)際還要看數(shù)據(jù)特征。如果數(shù)據(jù)里字符串特別多對象開銷會更大如果有大量shuffle內(nèi)存需求也會上升。我的經(jīng)驗(yàn)是先按估算值跑一次看Spark UI里的GC時間和spill情況再針對性調(diào)整。5.3 提交后常見報錯與排查提交到集群后最常見的報錯有這么幾類第一類ClassNotFoundException。通常是依賴沒打進(jìn)去。用Maven的shade插件打一個fat jar把所有依賴都包含進(jìn)去。注意Spark本身的依賴要設(shè)成provided因?yàn)榧荷弦呀?jīng)有了打進(jìn)去反而可能沖突。第二類Container killed by YARN for exceeding memory limits。這是Executor內(nèi)存超了?？赡苁菙?shù)據(jù)傾斜導(dǎo)致某個task處理的數(shù)據(jù)量特別大也可能是內(nèi)存配置本身就不夠。先看Spark UI里各task的處理數(shù)據(jù)量確認(rèn)是不是傾斜再決定是加內(nèi)存還是解決傾斜。第三類Shuffle fetch failed。通常是Executor掛了導(dǎo)致shuffle數(shù)據(jù)丟失。可以開啟spark.shuffle.service.enabled讓shuffle數(shù)據(jù)由外部服務(wù)管理Executor掛了也不影響。另外spark.network.timeout可以適當(dāng)調(diào)大避免網(wǎng)絡(luò)抖動導(dǎo)致的fetch失敗。第四類輸出目錄已存在。前面提過Spark不允許輸出到已存在的目錄?？梢栽诖a里加判斷或者每次提交前手動清理。6. 從案例到生產(chǎn)的思維轉(zhuǎn)變6.1 本地模式與集群模式的差異本地模式跑通不代表集群上沒問題。我總結(jié)了幾點(diǎn)差異都是實(shí)際踩過的第一文件路徑。本地用file:///集群上要用HDFS路徑。如果代碼里寫死了本地路徑提交上去必然報錯。建議把路徑作為參數(shù)傳入不要硬編碼。第二序列化。本地模式下很多對象不需要序列化就能傳遞集群模式下所有在Executor上執(zhí)行的對象都必須可序列化。如果代碼里用了自定義類記得實(shí)現(xiàn)Serializable接口。否則會報NotSerializableException而且報錯信息往往指向一個看起來無關(guān)的地方排查起來很費(fèi)勁。第三日志級別。本地跑的時候控制臺輸出很清晰集群上跑日志分散在各個Executor的container里需要去YARN的日志聚合里看。建議在代碼里把日志級別調(diào)成WARN減少不必要的輸出方便定位問題。6.2 代碼健壯性的幾個習(xí)慣寫Spark代碼有幾個習(xí)慣能幫你省很多事習(xí)慣一參數(shù)校驗(yàn)。輸入路徑、輸出路徑、TopN這些參數(shù)在main函數(shù)開頭就校驗(yàn)不合法直接退出別等到運(yùn)行到一半才報錯。習(xí)慣二資源關(guān)閉。spark.stop()要確保執(zhí)行最好放在finally塊里。雖然SparkSession不關(guān)也不會導(dǎo)致嚴(yán)重問題但長期運(yùn)行的服務(wù)里不關(guān)會占著資源。習(xí)慣三異常捕獲。對于可能失敗的操作比如讀寫HDFS加try-catch把異常信息打出來方便排查。但不要吞異常捕獲了要處理或者重新拋出。習(xí)慣四配置外置。把路徑、并行度、內(nèi)存這些配置抽到配置文件或者命令行參數(shù)里不要硬編碼。這樣換環(huán)境的時候不用改代碼重新提交就行。6.3 性能監(jiān)控與調(diào)優(yōu)的切入點(diǎn)作業(yè)跑起來之后怎么知道性能好不好看Spark UI。重點(diǎn)看幾個指標(biāo)Task的duration分布如果大部分task很快少數(shù)task特別慢說明有數(shù)據(jù)傾斜。GC時間如果GC時間占比超過10%說明內(nèi)存不夠需要加內(nèi)存或者減少緩存。Shuffle read/write如果shuffle數(shù)據(jù)量特別大考慮優(yōu)化算子減少shuffle。Spill如果出現(xiàn)spill說明內(nèi)存不足以容納當(dāng)前分區(qū)的數(shù)據(jù)需要加內(nèi)存或者增加分區(qū)數(shù)。調(diào)優(yōu)是一個迭代的過程沒有一勞永逸的配置。我的做法是先保證功能正確再根據(jù)監(jiān)控數(shù)據(jù)逐步優(yōu)化。每次只調(diào)一個參數(shù)觀察效果避免同時改多個參數(shù)導(dǎo)致無法定位問題。7. 一些容易被忽略的細(xì)節(jié)7.1 文本編碼問題讀取文本文件時默認(rèn)是UTF-8編碼。如果文件是GBK編碼中文會出現(xiàn)亂碼。解決辦法是在讀取時指定編碼或者先用工具轉(zhuǎn)成UTF-8。sc.textFile本身不提供編碼參數(shù)可以用sparkContext.hadoopConfiguration設(shè)置或者用wholeTextFiles讀取后手動解碼。7.2 空文件與空分區(qū)的處理如果輸入目錄里有空文件textFile會正常讀取但對應(yīng)的分區(qū)里沒有數(shù)據(jù)。這本身不是問題但如果后續(xù)有mapPartitions之類的操作要處理空迭代器的情況否則可能報錯。7.3 結(jié)果驗(yàn)證的方法跑完WordCount怎么驗(yàn)證結(jié)果對不對我的做法是先用一個小樣本數(shù)據(jù)手動算一遍預(yù)期結(jié)果然后跟Spark的輸出對比。確認(rèn)邏輯沒問題后再上全量數(shù)據(jù)。全量數(shù)據(jù)跑完后可以抽樣檢查幾個key的count跟原始數(shù)據(jù)里的出現(xiàn)次數(shù)對一下。對于TopKey驗(yàn)證方法是先跑完整的WordCount把結(jié)果按count排序取前N個跟TopKey的輸出對比。如果一致說明TopKey的邏輯沒問題。7.4 關(guān)于Scala的隱式轉(zhuǎn)換Scala的隱式轉(zhuǎn)換在Spark代碼里用得很多比如reduceByKey需要(K, V)類型的RDD而map返回的是普通RDD中間就涉及隱式轉(zhuǎn)換。如果import不對編譯會報錯。常見的import包括org.apache.spark.SparkContext._它包含了這些隱式轉(zhuǎn)換。Spark 2.x之后很多隱式轉(zhuǎn)換移到了org.apache.spark.rdd.RDD的伴生對象里import方式有變化需要注意版本差異。8. 從這兩個案例延伸出去WordCount和TopKey雖然簡單但它們覆蓋的算子組合可以延伸到很多場景。比如把單詞換成URL就是訪問量統(tǒng)計(jì)把單詞換成錯誤碼就是錯誤分布分析把count換成金額就是銷售額聚合。核心模式是一樣的拆分、映射、聚合、排序。掌握了這個模式再去看Spark的其他算子比如aggregateByKey、combineByKey、foldByKey會發(fā)現(xiàn)它們只是聚合方式不同底層邏輯是相通的。reduceByKey要求聚合前后的類型一致aggregateByKey允許聚合前后類型不同combineByKey則更靈活可以指定初始創(chuàng)建、分區(qū)內(nèi)合并、分區(qū)間合并三個函數(shù)。我在實(shí)際項(xiàng)目中用得最多的是reduceByKey和aggregateByKey。前者適合簡單的求和、計(jì)數(shù)后者適合求平均值、求TopN這類需要中間狀態(tài)的聚合。比如求每個用戶的平均消費(fèi)金額就可以用aggregateByKey初始值是(0.0, 0)分區(qū)內(nèi)合并是累加金額和次數(shù)分區(qū)間合并也是累加最后再相除。最后分享一個小技巧寫Spark代碼的時候盡量用reduceByKey替代groupByKey用mapPartitions替代map當(dāng)需要在每個分區(qū)內(nèi)做初始化操作時用filter盡早過濾掉不需要的數(shù)據(jù)。這些習(xí)慣能顯著提升作業(yè)性能尤其是在數(shù)據(jù)量大的時候。