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