1. 從一個看起來簡單的WordCount說起但凡接觸過分布式計(jì)算的人幾乎都繞不開WordCount這個案例。它就像編程語言里的Hello World看著簡單但真正動手寫的時候你會發(fā)現(xiàn)里面藏著不少門道。我見過太多人第一次跑WordCount代碼照著示例敲完提交上去結(jié)果要么報(bào)錯要么輸出一堆空文件要么結(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。這個必須對齊否則編譯的時候會報(bào)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é)果編譯報(bào)錯找不到類。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:///路徑測試提交到集群后報(bào)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 提交后常見報(bào)錯與排查提交到集群后最常見的報(bào)錯有這么幾類第一類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不允許輸出到已存在的目錄。可以在代碼里加判斷或者每次提交前手動清理。6. 從案例到生產(chǎn)的思維轉(zhuǎn)變6.1 本地模式與集群模式的差異本地模式跑通不代表集群上沒問題。我總結(jié)了幾點(diǎn)差異都是實(shí)際踩過的第一文件路徑。本地用file:///集群上要用HDFS路徑。如果代碼里寫死了本地路徑提交上去必然報(bào)錯。建議把路徑作為參數(shù)傳入不要硬編碼。第二序列化。本地模式下很多對象不需要序列化就能傳遞集群模式下所有在Executor上執(zhí)行的對象都必須可序列化。如果代碼里用了自定義類記得實(shí)現(xiàn)Serializable接口。否則會報(bào)NotSerializableException而且報(bào)錯信息往往指向一個看起來無關(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)行到一半才報(bào)錯。習(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)控?cái)?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之類的操作要處理空迭代器的情況否則可能報(bào)錯。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不對編譯會報(bào)錯。常見的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ù)量大的時候。