優(yōu))
1. 為什么foreachRDD是Spark Streaming里最容易翻車的算子剛接觸Spark Streaming那會兒我對foreachRDD的理解就是把每個(gè)批次的數(shù)據(jù)拿出來處理一下看起來跟RDD的foreach差不多能有多難結(jié)果第一次在生產(chǎn)環(huán)境用它往外部存儲寫數(shù)據(jù)就踩了個(gè)大坑——每個(gè)批次都新建一個(gè)連接跑了幾分鐘直接把連接池打滿了任務(wù)全線卡死。后來復(fù)盤才發(fā)現(xiàn)問題根本不在業(yè)務(wù)邏輯而在于我對foreachRDD的執(zhí)行位置和生命周期完全沒有概念。foreachRDD是Spark Streaming中唯一一個(gè)逃出DStream抽象、直接操作底層RDD的算子。它的定位非常特殊DStream本身是一系列RDD的序列而foreachRDD讓你在每個(gè)批次生成的RDD上執(zhí)行任意操作。這個(gè)任意就是雙刃劍——它給了你最大的靈活性也把資源管理的責(zé)任全部甩給了你。你可以用它寫數(shù)據(jù)庫、寫消息隊(duì)列、寫文件系統(tǒng)、調(diào)用外部接口但每一個(gè)操作都需要你自己考慮連接復(fù)用、序列化、冪等性、失敗重試這些事。這篇文章適合已經(jīng)寫過基礎(chǔ)Spark Streaming任務(wù)、但對foreachRDD的內(nèi)部機(jī)制和最佳實(shí)踐還不夠清晰的開發(fā)者。我會從執(zhí)行架構(gòu)講起把連接管理的坑、序列化的坑、冪等性的坑一個(gè)個(gè)拆開配上可以直接抄的代碼模板。如果你正在用foreachRDD往MySQL、HBase、Kafka或者Redis寫數(shù)據(jù)這篇內(nèi)容應(yīng)該能幫你少走不少彎路。2. foreachRDD在Spark Streaming執(zhí)行鏈路中的真實(shí)位置2.1 DStream的惰性本質(zhì)與foreachRDD的觸發(fā)角色要理解foreachRDD得先搞清楚DStream到底是什么。DStream并不是一個(gè)裝著數(shù)據(jù)的容器它更像一份生產(chǎn)計(jì)劃書——描述了數(shù)據(jù)從哪里來、經(jīng)過哪些變換、最終要到哪里去。真正承載數(shù)據(jù)的是每個(gè)批次間隔batch interval生成的一個(gè)個(gè)RDD。這些RDD按時(shí)間順序排列就構(gòu)成了DStream的底層結(jié)構(gòu)。Spark Streaming的調(diào)度器會按照batch interval周期性地觸發(fā)作業(yè)。每個(gè)周期內(nèi)DStream Graph中所有的轉(zhuǎn)換操作會被編譯成一個(gè)Job交給底層的Spark引擎執(zhí)行。而foreachRDD就是這個(gè)Graph的終點(diǎn)站——它是一個(gè)Output Operation只有Output Operation才會真正觸發(fā)Job的提交。沒有Output Operation的DStream Graph就是一張廢紙什么都不會執(zhí)行。這一點(diǎn)跟RDD的惰性求值是一個(gè)道理。你寫了一堆map、filter、reduceByKey只要沒有actionSpark什么都不會做。DStream也一樣foreachRDD就是那個(gè)action級別的存在。區(qū)別在于RDD的action觸發(fā)一次計(jì)算就結(jié)束了而foreachRDD會在每個(gè)batch interval被反復(fù)觸發(fā)直到StreamingContext停止。2.2 Driver端與Executor端的執(zhí)行邊界foreachRDD最容易被誤解的地方就是它的代碼到底在哪里執(zhí)行。很多人以為foreachRDD里面的代碼都在Executor上跑其實(shí)不是。foreachRDD的方法體本身是在Driver端執(zhí)行的它接收一個(gè)函數(shù)func: RDD[T] Unit這個(gè)函數(shù)在每個(gè)批次被調(diào)用一次調(diào)用發(fā)生在Driver上。但函數(shù)內(nèi)部對RDD的操作就不一樣了。如果你在foreachRDD里面寫了rdd.foreach(...)那么rdd.foreach里面的代碼才會被分發(fā)到Executor上執(zhí)行。如果你寫的是rdd.collect()數(shù)據(jù)會被拉到Driver端。如果你寫的是rdd.saveAsTextFile(...)寫操作由Executor完成。這個(gè)邊界搞不清楚就會犯一個(gè)經(jīng)典錯(cuò)誤在foreachRDD的方法體里直接創(chuàng)建數(shù)據(jù)庫連接。因?yàn)榉椒w在Driver上執(zhí)行你創(chuàng)建的連接對象在Driver上然后你試圖在rdd.foreach里用它——這個(gè)連接對象需要被序列化后發(fā)送到Executor而數(shù)據(jù)庫連接根本不可序列化直接報(bào)NotSerializableException。就算某些連接碰巧能序列化每個(gè)Executor上拿到的也是一個(gè)反序列化后的副本跟Driver上的原始連接沒有任何關(guān)系用完了也不會關(guān)閉連接泄漏是必然的。2.3 批次間隔與foreachRDD調(diào)用頻率的關(guān)系foreachRDD的調(diào)用頻率等于batch interval的倒數(shù)。如果你設(shè)置的batch interval是5秒那么foreachRDD的方法體每5秒被執(zhí)行一次里面的RDD操作也會每5秒觸發(fā)一次。這個(gè)頻率直接決定了你對外部系統(tǒng)的寫入壓力。我見過有同學(xué)把batch interval設(shè)成1秒然后在foreachRDD里往MySQL寫數(shù)據(jù)每個(gè)批次建一次連接、關(guān)一次連接。1秒一次連接創(chuàng)建和銷毀MySQL的wait_timeout還沒到連接數(shù)就已經(jīng)爆了。這不是MySQL的問題是batch interval和連接管理策略沒有匹配好。一般來說batch interval的設(shè)置要綜合考慮數(shù)據(jù)量、處理延遲要求和外部系統(tǒng)的承載能力。如果外部系統(tǒng)對連接創(chuàng)建比較敏感要么把batch interval調(diào)大要么在foreachRDD內(nèi)部做連接池復(fù)用。這兩個(gè)方向后面都會詳細(xì)講。3. 連接管理的三種寫法與它們的性能差距3.1 最直覺但最致命的寫法每個(gè)RDD建一次連接先看一段很多人一開始都會寫的代碼dstream.foreachRDD { rdd val connection createConnection() // 在Driver上執(zhí)行 rdd.foreach { record connection.write(record) // 連接被序列化到Executor必然失敗 } connection.close() }這段代碼有兩個(gè)致命問題。第一createConnection()在Driver上執(zhí)行得到的連接對象要隨rdd.foreach的閉包一起序列化發(fā)送到Executor絕大多數(shù)數(shù)據(jù)庫連接不支持序列化直接拋異常。第二即使某個(gè)連接庫支持序列化每個(gè)Executor上反序列化出來的連接是獨(dú)立的副本connection.close()在Driver上調(diào)用關(guān)的是Driver上的那個(gè)連接Executor上的連接永遠(yuǎn)不會被關(guān)閉。正確的做法是把連接的創(chuàng)建和關(guān)閉都放到Executor端dstream.foreachRDD { rdd rdd.foreachPartition { partition val connection createConnection() // 在Executor上執(zhí)行 partition.foreach { record connection.write(record) } connection.close() // 在Executor上關(guān)閉 } }用foreachPartition代替foreach連接的創(chuàng)建和關(guān)閉就都發(fā)生在Executor端了。每個(gè)分區(qū)一個(gè)連接而不是每條記錄一個(gè)連接性能提升是數(shù)量級的。但即便如此如果分區(qū)數(shù)很多每個(gè)批次仍然會創(chuàng)建大量連接。假設(shè)你的RDD有200個(gè)分區(qū)batch interval是10秒那么每10秒就會創(chuàng)建和銷毀200個(gè)連接一分鐘就是1200個(gè)。對于連接創(chuàng)建開銷大的數(shù)據(jù)庫來說這個(gè)壓力依然不小。3.2 連接池方案用單例對象管理連接生命周期連接池的核心思路是讓連接在Executor的JVM生命周期內(nèi)復(fù)用而不是每個(gè)批次都重新創(chuàng)建。實(shí)現(xiàn)方式是利用Scala的object單例特性——每個(gè)Executor的JVM里這個(gè)object只會被初始化一次里面的連接池也就只創(chuàng)建一次。object ConnectionPoolManager extends Serializable { transient private var pool: HikariDataSource _ def getPool(): HikariDataSource { if (pool null) { synchronized { if (pool null) { val config new HikariConfig() config.setJdbcUrl(jdbc:mysql://host:3306/db) config.setUsername(user) config.setPassword(pass) config.setMaximumPoolSize(10) config.setMinimumIdle(2) pool new HikariDataSource(config) } } } pool } }然后在foreachRDD里這樣用dstream.foreachRDD { rdd rdd.foreachPartition { partition val pool ConnectionPoolManager.getPool() val connection pool.getConnection() try { partition.foreach { record // 用connection寫數(shù)據(jù) } } finally { connection.close() // 歸還連接到池不是真正關(guān)閉 } } }這里有幾個(gè)關(guān)鍵點(diǎn)需要注意。transient注解讓pool字段不被序列化因?yàn)镃onnectionPoolManager這個(gè)對象本身會被序列化到Executor但連接池不應(yīng)該被序列化。getPool()方法里的雙重檢查鎖保證了線程安全因?yàn)橐粋€(gè)Executor上可能有多個(gè)Task并發(fā)執(zhí)行。connection.close()在HikariCP里是歸還連接到池不是物理關(guān)閉所以可以放心調(diào)用。連接池的大小需要根據(jù)Executor的核數(shù)和并發(fā)Task數(shù)來調(diào)整。如果每個(gè)Executor有4個(gè)核最多同時(shí)跑4個(gè)Task那么maximumPoolSize設(shè)成4到8就夠了。設(shè)太大反而會浪費(fèi)資源還可能把數(shù)據(jù)庫的連接數(shù)打滿。3.3 三種方案的量化對比方案連接創(chuàng)建頻率每批次連接數(shù)適用場景風(fēng)險(xiǎn)點(diǎn)每RDD建連接每批次一次1但不可用無序列化失敗、連接泄漏每分區(qū)建連接每批次每分區(qū)一次等于分區(qū)數(shù)低頻批次、小分區(qū)數(shù)連接創(chuàng)建開銷大連接池Executor生命周期一次等于并發(fā)Task數(shù)高頻批次、大分區(qū)數(shù)池大小配置不當(dāng)從實(shí)際經(jīng)驗(yàn)來看只要你的batch interval小于30秒或者分區(qū)數(shù)超過50就應(yīng)該用連接池方案。每分區(qū)建連接的方式只適合batch interval較大比如1分鐘以上、分區(qū)數(shù)較少比如10個(gè)以內(nèi)的場景。4. 序列化問題為什么你的連接對象總是報(bào)錯(cuò)4.1 閉包序列化的底層機(jī)制Spark在分發(fā)Task到Executor時(shí)會把Task對應(yīng)的閉包序列化成字節(jié)數(shù)組。閉包會捕獲它引用的外部變量這些變量必須支持序列化。foreachRDD的方法體在Driver上執(zhí)行但方法體內(nèi)部的rdd.foreach或rdd.foreachPartition的閉包會被序列化發(fā)送到Executor。如果你在foreachRDD的方法體里定義了一個(gè)連接對象然后在rdd.foreach里引用它這個(gè)連接對象就會被閉包捕獲進(jìn)而被序列化。數(shù)據(jù)庫連接、網(wǎng)絡(luò)連接、文件句柄這些資源都不支持序列化所以會直接報(bào)java.io.NotSerializableException。更隱蔽的情況是你定義了一個(gè)包含連接字段的類這個(gè)類實(shí)現(xiàn)了Serializable接口但連接字段本身不可序列化。這種情況下序列化會在運(yùn)行時(shí)失敗而且報(bào)錯(cuò)信息可能不會直接指向連接字段排查起來比較費(fèi)勁。4.2 transient注解的正確使用姿勢transient是Scala和Java都支持的注解作用是標(biāo)記某個(gè)字段不參與序列化。在Spark Streaming的場景里它通常用來標(biāo)記那些不應(yīng)該被序列化、需要在Executor端重新初始化的字段。class DataWriter extends Serializable { transient private var connection: Connection _ def write(records: Iterator[Record]): Unit { if (connection null) { connection DriverManager.getConnection(url, user, pass) } records.foreach { record // 用connection寫 } } }這里connection被標(biāo)記為transient序列化時(shí)會被跳過。到了Executor端反序列化后connection是null需要在第一次使用時(shí)重新創(chuàng)建。這種惰性初始化的模式在Spark里非常常見。但要注意transient只是讓字段不被序列化它不會自動幫你重新初始化。你必須在代碼里顯式判斷null并創(chuàng)建連接否則會報(bào)NullPointerException。而且如果多個(gè)Task共享同一個(gè)對象實(shí)例還需要考慮線程安全問題。4.3 一個(gè)真實(shí)的序列化報(bào)錯(cuò)排查過程有一次同事找我?guī)兔匆粋€(gè)任務(wù)報(bào)錯(cuò)信息是Task not serializable但代碼里看起來沒有任何明顯的不可序列化對象。我讓他把foreachRDD里的代碼逐行注釋最后定位到一行看起來人畜無害的代碼val logger LoggerFactory.getLogger(this.getClass) dstream.foreachRDD { rdd rdd.foreach { record logger.info(sProcessing $record) // 這里引用了logger } }問題出在logger上。LoggerFactory.getLogger返回的Logger對象內(nèi)部持有大量不可序列化的狀態(tài)被閉包捕獲后導(dǎo)致序列化失敗。解決辦法是在rdd.foreach內(nèi)部重新獲取Logger或者用transient標(biāo)記Logger字段。這個(gè)案例說明序列化問題不一定來自你顯式創(chuàng)建的連接對象任何被閉包捕獲的、不可序列化的對象都可能成為罪魁禍?zhǔn)?。排查的時(shí)候最有效的方法就是逐行注釋二分定位。5. 冪等性與Exactly-OnceforeachRDD繞不開的難題5.1 At-Least-Once語義下重復(fù)寫入的根源Spark Streaming默認(rèn)提供的是At-Least-Once語義。這意味著在發(fā)生故障恢復(fù)時(shí)某些批次的數(shù)據(jù)可能會被重復(fù)處理。對于foreachRDD來說重復(fù)處理就意味著重復(fù)寫入外部系統(tǒng)。重復(fù)寫入的根源在于Spark Streaming的Checkpoint機(jī)制和WALWrite-Ahead Log。當(dāng)Executor失敗時(shí)Driver會根據(jù)Checkpoint信息重新調(diào)度失敗的批次。如果失敗發(fā)生在數(shù)據(jù)已經(jīng)寫入外部系統(tǒng)之后、但Checkpoint還沒更新之前那么這個(gè)批次就會被重新執(zhí)行導(dǎo)致重復(fù)寫入。舉個(gè)例子你的任務(wù)從Kafka讀取數(shù)據(jù)用foreachRDD寫入MySQL。某個(gè)批次的數(shù)據(jù)已經(jīng)成功寫入了MySQL但在更新Checkpoint時(shí)Driver崩潰了?；謴?fù)后Driver從上一個(gè)Checkpoint點(diǎn)重新開始這個(gè)批次會被重新處理同樣的數(shù)據(jù)會被再次寫入MySQL。如果你的寫入操作是簡單的INSERTMySQL里就會出現(xiàn)重復(fù)記錄。5.2 用冪等寫入對抗重復(fù)數(shù)據(jù)解決重復(fù)寫入最直接的辦法是讓寫入操作本身具備冪等性。冪等的意思是同樣的操作執(zhí)行一次和執(zhí)行多次對外部系統(tǒng)的影響是一樣的。對于MySQL常見的冪等方案有三種。第一種是用INSERT ... ON DUPLICATE KEY UPDATE前提是表上有唯一鍵。第二種是先用REPLACE INTO效果類似。第三種是先DELETE再INSERT但這種方式在并發(fā)場景下可能有問題。INSERT INTO user_events (event_id, user_id, event_type, event_time) VALUES (?, ?, ?, ?) ON DUPLICATE KEY UPDATE user_id VALUES(user_id), event_type VALUES(event_type), event_time VALUES(event_time)對于HBase可以利用RowKey的唯一性用Put操作天然實(shí)現(xiàn)冪等——同樣的RowKey重復(fù)寫入后寫的會覆蓋先寫的結(jié)果是一樣的。對于Kafka可以啟用Producer的冪等性enable.idempotencetrueKafka會為每條消息分配序列號Broker端會去重。但要注意Kafka的冪等性只在單個(gè)Producer會話內(nèi)有效跨會話的重復(fù)還是需要業(yè)務(wù)層處理。5.3 事務(wù)性寫入與手動提交Offset的配合如果你需要更強(qiáng)的Exactly-Once保證就需要把數(shù)據(jù)寫入和Offset提交放在同一個(gè)事務(wù)里?；舅悸肥窃谔幚砻總€(gè)批次時(shí)先開啟事務(wù)寫入數(shù)據(jù)同時(shí)把Offset信息也寫入同一個(gè)事務(wù)最后提交事務(wù)。如果事務(wù)失敗Offset不會提交批次會被重新處理。dstream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges rdd.foreachPartition { partition val connection getConnection() connection.setAutoCommit(false) try { partition.foreach { record // 寫入業(yè)務(wù)數(shù)據(jù) } // 在同一個(gè)事務(wù)里寫入Offset offsetRanges.foreach { range writeOffset(connection, range) } connection.commit() } catch { case e: Exception connection.rollback() throw e } finally { connection.close() } } }這種方案的關(guān)鍵在于Offset的存儲位置必須和業(yè)務(wù)數(shù)據(jù)在同一個(gè)數(shù)據(jù)庫、同一個(gè)事務(wù)里。如果Offset存在ZooKeeper或Kafka里就無法和業(yè)務(wù)數(shù)據(jù)寫入構(gòu)成原子操作Exactly-Once就無從談起。需要提醒的是這種方案會顯著增加寫入延遲因?yàn)槊總€(gè)批次都要等事務(wù)提交完成。而且如果事務(wù)回滾整個(gè)批次的數(shù)據(jù)都要重寫對數(shù)據(jù)庫的壓力會成倍增加。在實(shí)際項(xiàng)目中我通常只在數(shù)據(jù)準(zhǔn)確性要求極高的場景比如金融交易記錄才用這種方案普通場景用冪等寫入就夠了。6. 性能調(diào)優(yōu)讓foreachRDD不再成為瓶頸6.1 批量寫入代替逐條寫入逐條寫入是foreachRDD最常見的性能殺手。每寫一條記錄就發(fā)一次網(wǎng)絡(luò)請求網(wǎng)絡(luò)往返時(shí)間RTT累積起來非?？捎^。假設(shè)RTT是1毫秒寫10000條記錄就是10秒這還沒算數(shù)據(jù)庫端的處理時(shí)間。批量寫入的思路是把多條記錄攢成一個(gè)批次一次性發(fā)給外部系統(tǒng)。對于JDBC可以用addBatch()和executeBatch()rdd.foreachPartition { partition val connection getConnection() val statement connection.prepareStatement(INSERT INTO ...) var count 0 partition.foreach { record statement.setString(1, record.field1) statement.setInt(2, record.field2) statement.addBatch() count 1 if (count % 500 0) { statement.executeBatch() count 0 } } if (count 0) { statement.executeBatch() } statement.close() connection.close() }批量大小需要根據(jù)記錄大小和數(shù)據(jù)庫配置來調(diào)整。一般來說500到1000條一批比較合適。太小了效果不明顯太大了可能超過數(shù)據(jù)庫的max_allowed_packet限制。6.2 控制分區(qū)數(shù)避免連接爆炸RDD的分區(qū)數(shù)直接決定了foreachPartition的并發(fā)度也決定了同時(shí)創(chuàng)建的連接數(shù)。如果分區(qū)數(shù)是500每個(gè)分區(qū)創(chuàng)建一個(gè)連接那么瞬間就會有500個(gè)連接打到數(shù)據(jù)庫上。大多數(shù)數(shù)據(jù)庫的默認(rèn)最大連接數(shù)是100到200直接就被打滿了?？刂品謪^(qū)數(shù)有幾種方式。如果數(shù)據(jù)源是Kafka可以用directStream的repartition或者調(diào)整spark.streaming.kafka.maxRatePerPartition來控制。如果是receive方式可以調(diào)整spark.streaming.blockInterval來間接影響分區(qū)數(shù)。更直接的方式是在foreachRDD之前調(diào)用rdd.repartition(n)或rdd.coalesce(n)來顯式控制分區(qū)數(shù)。但要注意repartition會觸發(fā)Shuffle有額外的開銷coalesce不觸發(fā)Shuffle但只能減少分區(qū)數(shù)不能增加。我的經(jīng)驗(yàn)是分區(qū)數(shù)控制在Executor總核數(shù)的2到3倍比較合適。比如你有10個(gè)Executor每個(gè)4核總共40核那么分區(qū)數(shù)設(shè)在80到120之間。這樣既能保證并發(fā)度又不會創(chuàng)建過多連接。6.3 異步寫入與背壓機(jī)制的取舍如果外部系統(tǒng)的寫入延遲比較高同步寫入會拖慢整個(gè)批次的處理速度進(jìn)而導(dǎo)致批次堆積。這時(shí)候可以考慮異步寫入——把數(shù)據(jù)放到一個(gè)隊(duì)列里由單獨(dú)的線程池負(fù)責(zé)寫入foreachRDD只負(fù)責(zé)入隊(duì)不等待寫入完成。object AsyncWriter { private val executor Executors.newFixedThreadPool(10) private val queue new LinkedBlockingQueue[Record](10000) def start(): Unit { (1 to 10).foreach { _ executor.submit(new Runnable { override def run(): Unit { while (true) { val record queue.take() // 寫入外部系統(tǒng) } } }) } } def write(record: Record): Unit { queue.put(record) // 隊(duì)列滿時(shí)會阻塞 } }異步寫入能顯著提升吞吐量但會帶來兩個(gè)問題。第一如果隊(duì)列滿了put操作會阻塞反而拖慢處理速度。第二異步寫入無法保證數(shù)據(jù)在批次完成時(shí)已經(jīng)落盤如果任務(wù)在這期間崩潰隊(duì)列里的數(shù)據(jù)就丟了。Spark Streaming本身有背壓機(jī)制spark.streaming.backpressure.enabledtrue可以根據(jù)處理速度動態(tài)調(diào)整數(shù)據(jù)攝入速率。但背壓只控制攝入不控制寫出。如果寫出是瓶頸背壓會導(dǎo)致攝入速率被壓低整體吞吐量上不去。這種情況下要么優(yōu)化寫出邏輯要么增加寫出并發(fā)度單純靠背壓解決不了問題。7. 那些文檔里不會寫的實(shí)戰(zhàn)經(jīng)驗(yàn)7.1 Checkpoint目錄的清理與版本兼容用了foreachRDD的任務(wù)通常都需要開啟Checkpoint因?yàn)橐４鍻ffset和DStream Graph的元信息。但Checkpoint目錄會隨著時(shí)間不斷增長如果不清理磁盤很快就會被占滿。Spark Streaming本身沒有提供自動清理Checkpoint的機(jī)制需要手動處理。常見的做法是寫一個(gè)定時(shí)任務(wù)定期刪除過期的Checkpoint文件。但要注意不能刪除正在使用的Checkpoint否則任務(wù)恢復(fù)時(shí)會失敗。另一個(gè)坑是Checkpoint的版本兼容性。如果你修改了DStream Graph的結(jié)構(gòu)比如增加了一個(gè)轉(zhuǎn)換操作從舊Checkpoint恢復(fù)時(shí)會報(bào)錯(cuò)因?yàn)樾屡fGraph不匹配。這種情況下只能刪除Checkpoint重新開始或者用spark.streaming.stopGracefullyOnShutdown優(yōu)雅停止后手動遷移。我的建議是如果業(yè)務(wù)邏輯可能頻繁變更盡量不要依賴Checkpoint恢復(fù)而是用外部存儲如ZooKeeper或數(shù)據(jù)庫來保存Offset每次啟動時(shí)從外部存儲讀取Offset重新構(gòu)建DStream。這樣DStream Graph的變更就不會受Checkpoint限制。7.2 foreachRDD內(nèi)部的異常處理策略foreachRDD內(nèi)部的異常如果沒被捕獲會導(dǎo)致整個(gè)批次失敗進(jìn)而觸發(fā)重試。如果異常是持續(xù)性的比如數(shù)據(jù)庫連不上重試會一直失敗任務(wù)會陷入死循環(huán)。合理的做法是在foreachRDD內(nèi)部做細(xì)粒度的異常處理。對于可恢復(fù)的異常如網(wǎng)絡(luò)抖動可以重試幾次對于不可恢復(fù)的異常如數(shù)據(jù)格式錯(cuò)誤應(yīng)該記錄日志并跳過而不是讓整個(gè)批次失敗。rdd.foreachPartition { partition partition.foreach { record try { // 寫入邏輯 } catch { case e: RetryableException // 重試邏輯 case e: Exception // 記錄到死信隊(duì)列或日志 logger.error(sFailed to process $record, e) } } }但要注意如果異常被吞掉了Spark Streaming會認(rèn)為批次處理成功Offset會正常提交。那些被跳過的數(shù)據(jù)就永久丟失了。所以跳過數(shù)據(jù)的前提是你已經(jīng)把失敗的數(shù)據(jù)記錄到了某個(gè)地方后續(xù)可以人工補(bǔ)償。7.3 本地測試與生產(chǎn)環(huán)境的差異本地測試foreachRDD時(shí)通常用local[*]模式所有Task在同一個(gè)JVM里跑連接池、單例對象的行為和生產(chǎn)環(huán)境完全不同。本地測試通過不代表生產(chǎn)環(huán)境沒問題。我踩過的一個(gè)坑是本地測試時(shí)連接池大小設(shè)成10跑得好好的到了生產(chǎn)環(huán)境每個(gè)Executor的JVM里都有一個(gè)連接池10個(gè)Executor就是100個(gè)連接直接把數(shù)據(jù)庫打滿了。本地測試時(shí)只有一個(gè)JVM連接池只有一份所以看不出問題。另一個(gè)差異是序列化。本地模式下閉包可能不需要真正序列化因?yàn)槎荚谕粋€(gè)JVM里所以一些序列化問題在本地不會暴露。到了集群模式閉包必須序列化問題才會顯現(xiàn)。所以foreachRDD相關(guān)的代碼一定要在真實(shí)集群環(huán)境上做集成測試至少要用yarn-client或standalone模式跑一遍才能發(fā)現(xiàn)連接池、序列化、并發(fā)這些方面的問題。7.4 監(jiān)控與告警別等任務(wù)掛了才發(fā)現(xiàn)foreachRDD的寫入失敗往往不會立即導(dǎo)致任務(wù)崩潰而是表現(xiàn)為批次處理時(shí)間逐漸變長、數(shù)據(jù)延遲越來越大。如果沒有監(jiān)控可能等到數(shù)據(jù)延遲幾小時(shí)才發(fā)現(xiàn)問題。需要監(jiān)控的指標(biāo)包括批次處理時(shí)間Batch Processing Time、調(diào)度延遲Scheduling Delay、每個(gè)批次的記錄數(shù)、寫入失敗率。Spark Streaming的Web UI提供了前兩個(gè)指標(biāo)后兩個(gè)需要自己在代碼里埋點(diǎn)。dstream.foreachRDD { rdd val startTime System.currentTimeMillis() val count rdd.count() rdd.foreachPartition { partition // 寫入邏輯 } val endTime System.currentTimeMillis() metricsReporter.report(write_latency, endTime - startTime) metricsReporter.report(record_count, count) }這些指標(biāo)可以推到Prometheus或Graphite配上告警規(guī)則。比如批次處理時(shí)間連續(xù)5分鐘超過batch interval的80%就觸發(fā)告警。這樣在問題惡化之前就能介入處理。8. 一個(gè)完整的生產(chǎn)級foreachRDD模板把前面講的要點(diǎn)整合起來下面是一個(gè)可以直接參考的生產(chǎn)級模板。這個(gè)模板包含了連接池管理、批量寫入、異常處理、冪等寫入和監(jiān)控埋點(diǎn)。object ProductionForeachRDDTemplate { def main(args: Array[String]): Unit { val conf new SparkConf() .setAppName(ProductionForeachRDD) .set(spark.streaming.backpressure.enabled, true) .set(spark.streaming.stopGracefullyOnShutdown, true) val ssc new StreamingContext(conf, Seconds(10)) ssc.checkpoint(hdfs://checkpoint/path) val kafkaParams Map( bootstrap.servers - kafka:9092, group.id - group1, auto.offset.reset - latest, enable.auto.commit - false ) val topics Set(topic1) val stream KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams) ) stream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges if (!rdd.isEmpty()) { val startTime System.currentTimeMillis() var totalCount 0L var failCount 0L rdd.foreachPartition { partition val pool ConnectionPoolManager.getPool() val connection pool.getConnection() connection.setAutoCommit(false) val statement connection.prepareStatement( INSERT INTO events (id, data, ts) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE data VALUES(data) ) var batchCount 0 try { partition.foreach { record try { val parsed parseRecord(record) statement.setString(1, parsed.id) statement.setString(2, parsed.data) statement.setLong(3, parsed.ts) statement.addBatch() batchCount 1 totalCount 1 if (batchCount % 500 0) { statement.executeBatch() connection.commit() batchCount 0 } } catch { case e: Exception failCount 1 logger.error(sFailed to process record: $record, e) } } if (batchCount 0) { statement.executeBatch() connection.commit() } } catch { case e: Exception connection.rollback() throw e } finally { statement.close() connection.close() } } // 提交Offset stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) val endTime System.currentTimeMillis() metricsReporter.report(batch_latency, endTime - startTime) metricsReporter.report(total_count, totalCount) metricsReporter.report(fail_count, failCount) } } ssc.start() ssc.awaitTermination() } }這個(gè)模板里有幾個(gè)設(shè)計(jì)決策值得說明。用createDirectStream而不是createStream是為了拿到OffsetRanges方便手動提交Offset。enable.auto.commit設(shè)為false避免Kafka自動提交Offset導(dǎo)致數(shù)據(jù)丟失。ON DUPLICATE KEY UPDATE保證了冪等性。批量大小設(shè)為500是經(jīng)驗(yàn)值可以根據(jù)實(shí)際情況調(diào)整。異常處理分兩層內(nèi)層捕獲單條記錄的異常避免一條壞數(shù)據(jù)導(dǎo)致整個(gè)批次失敗外層捕獲批量提交的異?；貪L事務(wù)并拋出讓Spark Streaming重試整個(gè)批次。連接池的配置需要根據(jù)Executor數(shù)量和數(shù)據(jù)庫承載能力來調(diào)整。如果每個(gè)Executor的maximumPoolSize是5有20個(gè)Executor那么總連接數(shù)是100。要確保數(shù)據(jù)庫的max_connections大于這個(gè)值并留出余量給其他應(yīng)用。監(jiān)控埋點(diǎn)建議推到外部監(jiān)控系統(tǒng)不要只打日志。日志在分布式環(huán)境下分散在各個(gè)Executor上排查問題很不方便。推到Prometheus或類似的系統(tǒng)可以統(tǒng)一查看和告警。這套模板我在多個(gè)生產(chǎn)任務(wù)里用過穩(wěn)定性不錯(cuò)。但每個(gè)項(xiàng)目的數(shù)據(jù)庫、數(shù)據(jù)量、延遲要求都不一樣具體參數(shù)需要根據(jù)實(shí)際情況調(diào)整。比如批量大小數(shù)據(jù)記錄大的時(shí)候要調(diào)小記錄小的時(shí)候可以調(diào)大。連接池大小也是數(shù)據(jù)庫性能好可以多開一些性能一般就要控制。最后說一個(gè)容易被忽略的點(diǎn)foreachRDD里的代碼在Driver上執(zhí)行的部分如果耗時(shí)太長比如做了復(fù)雜的計(jì)算會阻塞Driver的調(diào)度線程導(dǎo)致后續(xù)批次無法及時(shí)提交。所以foreachRDD方法體里盡量只做輕量的操作重活都放到rdd.foreachPartition里讓Executor去干。這個(gè)邊界意識是用好foreachRDD的關(guān)鍵。 Spark Streaming中foreachRDD的避坑指南：連接管理與性能調(diào)優(yōu)
1. 為什么foreachRDD是Spark Streaming里最容易翻車的算子剛接觸Spark Streaming那會兒我對foreachRDD的理解就是把每個(gè)批次的數(shù)據(jù)拿出來處理一下看起來跟RDD的foreach差不多能有多難結(jié)果第一次在生產(chǎn)環(huán)境用它往外部存儲寫數(shù)據(jù)就踩了個(gè)大坑——每個(gè)批次都新建一個(gè)連接跑了幾分鐘直接把連接池打滿了任務(wù)全線卡死。后來復(fù)盤才發(fā)現(xiàn)問題根本不在業(yè)務(wù)邏輯而在于我對foreachRDD的執(zhí)行位置和生命周期完全沒有概念。foreachRDD是Spark Streaming中唯一一個(gè)逃出DStream抽象、直接操作底層RDD的算子。它的定位非常特殊DStream本身是一系列RDD的序列而foreachRDD讓你在每個(gè)批次生成的RDD上執(zhí)行任意操作。這個(gè)任意就是雙刃劍——它給了你最大的靈活性也把資源管理的責(zé)任全部甩給了你。你可以用它寫數(shù)據(jù)庫、寫消息隊(duì)列、寫文件系統(tǒng)、調(diào)用外部接口但每一個(gè)操作都需要你自己考慮連接復(fù)用、序列化、冪等性、失敗重試這些事。這篇文章適合已經(jīng)寫過基礎(chǔ)Spark Streaming任務(wù)、但對foreachRDD的內(nèi)部機(jī)制和最佳實(shí)踐還不夠清晰的開發(fā)者。我會從執(zhí)行架構(gòu)講起把連接管理的坑、序列化的坑、冪等性的坑一個(gè)個(gè)拆開配上可以直接抄的代碼模板。如果你正在用foreachRDD往MySQL、HBase、Kafka或者Redis寫數(shù)據(jù)這篇內(nèi)容應(yīng)該能幫你少走不少彎路。2. foreachRDD在Spark Streaming執(zhí)行鏈路中的真實(shí)位置2.1 DStream的惰性本質(zhì)與foreachRDD的觸發(fā)角色要理解foreachRDD得先搞清楚DStream到底是什么。DStream并不是一個(gè)裝著數(shù)據(jù)的容器它更像一份生產(chǎn)計(jì)劃書——描述了數(shù)據(jù)從哪里來、經(jīng)過哪些變換、最終要到哪里去。真正承載數(shù)據(jù)的是每個(gè)批次間隔batch interval生成的一個(gè)個(gè)RDD。這些RDD按時(shí)間順序排列就構(gòu)成了DStream的底層結(jié)構(gòu)。Spark Streaming的調(diào)度器會按照batch interval周期性地觸發(fā)作業(yè)。每個(gè)周期內(nèi)DStream Graph中所有的轉(zhuǎn)換操作會被編譯成一個(gè)Job交給底層的Spark引擎執(zhí)行。而foreachRDD就是這個(gè)Graph的終點(diǎn)站——它是一個(gè)Output Operation只有Output Operation才會真正觸發(fā)Job的提交。沒有Output Operation的DStream Graph就是一張廢紙什么都不會執(zhí)行。這一點(diǎn)跟RDD的惰性求值是一個(gè)道理。你寫了一堆map、filter、reduceByKey只要沒有actionSpark什么都不會做。DStream也一樣foreachRDD就是那個(gè)action級別的存在。區(qū)別在于RDD的action觸發(fā)一次計(jì)算就結(jié)束了而foreachRDD會在每個(gè)batch interval被反復(fù)觸發(fā)直到StreamingContext停止。2.2 Driver端與Executor端的執(zhí)行邊界foreachRDD最容易被誤解的地方就是它的代碼到底在哪里執(zhí)行。很多人以為foreachRDD里面的代碼都在Executor上跑其實(shí)不是。foreachRDD的方法體本身是在Driver端執(zhí)行的它接收一個(gè)函數(shù)func: RDD[T] Unit這個(gè)函數(shù)在每個(gè)批次被調(diào)用一次調(diào)用發(fā)生在Driver上。但函數(shù)內(nèi)部對RDD的操作就不一樣了。如果你在foreachRDD里面寫了rdd.foreach(...)那么rdd.foreach里面的代碼才會被分發(fā)到Executor上執(zhí)行。如果你寫的是rdd.collect()數(shù)據(jù)會被拉到Driver端。如果你寫的是rdd.saveAsTextFile(...)寫操作由Executor完成。這個(gè)邊界搞不清楚就會犯一個(gè)經(jīng)典錯(cuò)誤在foreachRDD的方法體里直接創(chuàng)建數(shù)據(jù)庫連接。因?yàn)榉椒w在Driver上執(zhí)行你創(chuàng)建的連接對象在Driver上然后你試圖在rdd.foreach里用它——這個(gè)連接對象需要被序列化后發(fā)送到Executor而數(shù)據(jù)庫連接根本不可序列化直接報(bào)NotSerializableException。就算某些連接碰巧能序列化每個(gè)Executor上拿到的也是一個(gè)反序列化后的副本跟Driver上的原始連接沒有任何關(guān)系用完了也不會關(guān)閉連接泄漏是必然的。2.3 批次間隔與foreachRDD調(diào)用頻率的關(guān)系foreachRDD的調(diào)用頻率等于batch interval的倒數(shù)。如果你設(shè)置的batch interval是5秒那么foreachRDD的方法體每5秒被執(zhí)行一次里面的RDD操作也會每5秒觸發(fā)一次。這個(gè)頻率直接決定了你對外部系統(tǒng)的寫入壓力。我見過有同學(xué)把batch interval設(shè)成1秒然后在foreachRDD里往MySQL寫數(shù)據(jù)每個(gè)批次建一次連接、關(guān)一次連接。1秒一次連接創(chuàng)建和銷毀MySQL的wait_timeout還沒到連接數(shù)就已經(jīng)爆了。這不是MySQL的問題是batch interval和連接管理策略沒有匹配好。一般來說batch interval的設(shè)置要綜合考慮數(shù)據(jù)量、處理延遲要求和外部系統(tǒng)的承載能力。如果外部系統(tǒng)對連接創(chuàng)建比較敏感要么把batch interval調(diào)大要么在foreachRDD內(nèi)部做連接池復(fù)用。這兩個(gè)方向后面都會詳細(xì)講。3. 連接管理的三種寫法與它們的性能差距3.1 最直覺但最致命的寫法每個(gè)RDD建一次連接先看一段很多人一開始都會寫的代碼dstream.foreachRDD { rdd val connection createConnection() // 在Driver上執(zhí)行 rdd.foreach { record connection.write(record) // 連接被序列化到Executor必然失敗 } connection.close() }這段代碼有兩個(gè)致命問題。第一createConnection()在Driver上執(zhí)行得到的連接對象要隨rdd.foreach的閉包一起序列化發(fā)送到Executor絕大多數(shù)數(shù)據(jù)庫連接不支持序列化直接拋異常。第二即使某個(gè)連接庫支持序列化每個(gè)Executor上反序列化出來的連接是獨(dú)立的副本connection.close()在Driver上調(diào)用關(guān)的是Driver上的那個(gè)連接Executor上的連接永遠(yuǎn)不會被關(guān)閉。正確的做法是把連接的創(chuàng)建和關(guān)閉都放到Executor端dstream.foreachRDD { rdd rdd.foreachPartition { partition val connection createConnection() // 在Executor上執(zhí)行 partition.foreach { record connection.write(record) } connection.close() // 在Executor上關(guān)閉 } }用foreachPartition代替foreach連接的創(chuàng)建和關(guān)閉就都發(fā)生在Executor端了。每個(gè)分區(qū)一個(gè)連接而不是每條記錄一個(gè)連接性能提升是數(shù)量級的。但即便如此如果分區(qū)數(shù)很多每個(gè)批次仍然會創(chuàng)建大量連接。假設(shè)你的RDD有200個(gè)分區(qū)batch interval是10秒那么每10秒就會創(chuàng)建和銷毀200個(gè)連接一分鐘就是1200個(gè)。對于連接創(chuàng)建開銷大的數(shù)據(jù)庫來說這個(gè)壓力依然不小。3.2 連接池方案用單例對象管理連接生命周期連接池的核心思路是讓連接在Executor的JVM生命周期內(nèi)復(fù)用而不是每個(gè)批次都重新創(chuàng)建。實(shí)現(xiàn)方式是利用Scala的object單例特性——每個(gè)Executor的JVM里這個(gè)object只會被初始化一次里面的連接池也就只創(chuàng)建一次。object ConnectionPoolManager extends Serializable { transient private var pool: HikariDataSource _ def getPool(): HikariDataSource { if (pool null) { synchronized { if (pool null) { val config new HikariConfig() config.setJdbcUrl(jdbc:mysql://host:3306/db) config.setUsername(user) config.setPassword(pass) config.setMaximumPoolSize(10) config.setMinimumIdle(2) pool new HikariDataSource(config) } } } pool } }然后在foreachRDD里這樣用dstream.foreachRDD { rdd rdd.foreachPartition { partition val pool ConnectionPoolManager.getPool() val connection pool.getConnection() try { partition.foreach { record // 用connection寫數(shù)據(jù) } } finally { connection.close() // 歸還連接到池不是真正關(guān)閉 } } }這里有幾個(gè)關(guān)鍵點(diǎn)需要注意。transient注解讓pool字段不被序列化因?yàn)镃onnectionPoolManager這個(gè)對象本身會被序列化到Executor但連接池不應(yīng)該被序列化。getPool()方法里的雙重檢查鎖保證了線程安全因?yàn)橐粋€(gè)Executor上可能有多個(gè)Task并發(fā)執(zhí)行。connection.close()在HikariCP里是歸還連接到池不是物理關(guān)閉所以可以放心調(diào)用。連接池的大小需要根據(jù)Executor的核數(shù)和并發(fā)Task數(shù)來調(diào)整。如果每個(gè)Executor有4個(gè)核最多同時(shí)跑4個(gè)Task那么maximumPoolSize設(shè)成4到8就夠了。設(shè)太大反而會浪費(fèi)資源還可能把數(shù)據(jù)庫的連接數(shù)打滿。3.3 三種方案的量化對比方案連接創(chuàng)建頻率每批次連接數(shù)適用場景風(fēng)險(xiǎn)點(diǎn)每RDD建連接每批次一次1但不可用無序列化失敗、連接泄漏每分區(qū)建連接每批次每分區(qū)一次等于分區(qū)數(shù)低頻批次、小分區(qū)數(shù)連接創(chuàng)建開銷大連接池Executor生命周期一次等于并發(fā)Task數(shù)高頻批次、大分區(qū)數(shù)池大小配置不當(dāng)從實(shí)際經(jīng)驗(yàn)來看只要你的batch interval小于30秒或者分區(qū)數(shù)超過50就應(yīng)該用連接池方案。每分區(qū)建連接的方式只適合batch interval較大比如1分鐘以上、分區(qū)數(shù)較少比如10個(gè)以內(nèi)的場景。4. 序列化問題為什么你的連接對象總是報(bào)錯(cuò)4.1 閉包序列化的底層機(jī)制Spark在分發(fā)Task到Executor時(shí)會把Task對應(yīng)的閉包序列化成字節(jié)數(shù)組。閉包會捕獲它引用的外部變量這些變量必須支持序列化。foreachRDD的方法體在Driver上執(zhí)行但方法體內(nèi)部的rdd.foreach或rdd.foreachPartition的閉包會被序列化發(fā)送到Executor。如果你在foreachRDD的方法體里定義了一個(gè)連接對象然后在rdd.foreach里引用它這個(gè)連接對象就會被閉包捕獲進(jìn)而被序列化。數(shù)據(jù)庫連接、網(wǎng)絡(luò)連接、文件句柄這些資源都不支持序列化所以會直接報(bào)java.io.NotSerializableException。更隱蔽的情況是你定義了一個(gè)包含連接字段的類這個(gè)類實(shí)現(xiàn)了Serializable接口但連接字段本身不可序列化。這種情況下序列化會在運(yùn)行時(shí)失敗而且報(bào)錯(cuò)信息可能不會直接指向連接字段排查起來比較費(fèi)勁。4.2 transient注解的正確使用姿勢transient是Scala和Java都支持的注解作用是標(biāo)記某個(gè)字段不參與序列化。在Spark Streaming的場景里它通常用來標(biāo)記那些不應(yīng)該被序列化、需要在Executor端重新初始化的字段。class DataWriter extends Serializable { transient private var connection: Connection _ def write(records: Iterator[Record]): Unit { if (connection null) { connection DriverManager.getConnection(url, user, pass) } records.foreach { record // 用connection寫 } } }這里connection被標(biāo)記為transient序列化時(shí)會被跳過。到了Executor端反序列化后connection是null需要在第一次使用時(shí)重新創(chuàng)建。這種惰性初始化的模式在Spark里非常常見。但要注意transient只是讓字段不被序列化它不會自動幫你重新初始化。你必須在代碼里顯式判斷null并創(chuàng)建連接否則會報(bào)NullPointerException。而且如果多個(gè)Task共享同一個(gè)對象實(shí)例還需要考慮線程安全問題。4.3 一個(gè)真實(shí)的序列化報(bào)錯(cuò)排查過程有一次同事找我?guī)兔匆粋€(gè)任務(wù)報(bào)錯(cuò)信息是Task not serializable但代碼里看起來沒有任何明顯的不可序列化對象。我讓他把foreachRDD里的代碼逐行注釋最后定位到一行看起來人畜無害的代碼val logger LoggerFactory.getLogger(this.getClass) dstream.foreachRDD { rdd rdd.foreach { record logger.info(sProcessing $record) // 這里引用了logger } }問題出在logger上。LoggerFactory.getLogger返回的Logger對象內(nèi)部持有大量不可序列化的狀態(tài)被閉包捕獲后導(dǎo)致序列化失敗。解決辦法是在rdd.foreach內(nèi)部重新獲取Logger或者用transient標(biāo)記Logger字段。這個(gè)案例說明序列化問題不一定來自你顯式創(chuàng)建的連接對象任何被閉包捕獲的、不可序列化的對象都可能成為罪魁禍?zhǔn)?。排查的時(shí)候最有效的方法就是逐行注釋二分定位。5. 冪等性與Exactly-OnceforeachRDD繞不開的難題5.1 At-Least-Once語義下重復(fù)寫入的根源Spark Streaming默認(rèn)提供的是At-Least-Once語義。這意味著在發(fā)生故障恢復(fù)時(shí)某些批次的數(shù)據(jù)可能會被重復(fù)處理。對于foreachRDD來說重復(fù)處理就意味著重復(fù)寫入外部系統(tǒng)。重復(fù)寫入的根源在于Spark Streaming的Checkpoint機(jī)制和WALWrite-Ahead Log。當(dāng)Executor失敗時(shí)Driver會根據(jù)Checkpoint信息重新調(diào)度失敗的批次。如果失敗發(fā)生在數(shù)據(jù)已經(jīng)寫入外部系統(tǒng)之后、但Checkpoint還沒更新之前那么這個(gè)批次就會被重新執(zhí)行導(dǎo)致重復(fù)寫入。舉個(gè)例子你的任務(wù)從Kafka讀取數(shù)據(jù)用foreachRDD寫入MySQL。某個(gè)批次的數(shù)據(jù)已經(jīng)成功寫入了MySQL但在更新Checkpoint時(shí)Driver崩潰了?；謴?fù)后Driver從上一個(gè)Checkpoint點(diǎn)重新開始這個(gè)批次會被重新處理同樣的數(shù)據(jù)會被再次寫入MySQL。如果你的寫入操作是簡單的INSERTMySQL里就會出現(xiàn)重復(fù)記錄。5.2 用冪等寫入對抗重復(fù)數(shù)據(jù)解決重復(fù)寫入最直接的辦法是讓寫入操作本身具備冪等性。冪等的意思是同樣的操作執(zhí)行一次和執(zhí)行多次對外部系統(tǒng)的影響是一樣的。對于MySQL常見的冪等方案有三種。第一種是用INSERT ... ON DUPLICATE KEY UPDATE前提是表上有唯一鍵。第二種是先用REPLACE INTO效果類似。第三種是先DELETE再INSERT但這種方式在并發(fā)場景下可能有問題。INSERT INTO user_events (event_id, user_id, event_type, event_time) VALUES (?, ?, ?, ?) ON DUPLICATE KEY UPDATE user_id VALUES(user_id), event_type VALUES(event_type), event_time VALUES(event_time)對于HBase可以利用RowKey的唯一性用Put操作天然實(shí)現(xiàn)冪等——同樣的RowKey重復(fù)寫入后寫的會覆蓋先寫的結(jié)果是一樣的。對于Kafka可以啟用Producer的冪等性enable.idempotencetrueKafka會為每條消息分配序列號Broker端會去重。但要注意Kafka的冪等性只在單個(gè)Producer會話內(nèi)有效跨會話的重復(fù)還是需要業(yè)務(wù)層處理。5.3 事務(wù)性寫入與手動提交Offset的配合如果你需要更強(qiáng)的Exactly-Once保證就需要把數(shù)據(jù)寫入和Offset提交放在同一個(gè)事務(wù)里?；舅悸肥窃谔幚砻總€(gè)批次時(shí)先開啟事務(wù)寫入數(shù)據(jù)同時(shí)把Offset信息也寫入同一個(gè)事務(wù)最后提交事務(wù)。如果事務(wù)失敗Offset不會提交批次會被重新處理。dstream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges rdd.foreachPartition { partition val connection getConnection() connection.setAutoCommit(false) try { partition.foreach { record // 寫入業(yè)務(wù)數(shù)據(jù) } // 在同一個(gè)事務(wù)里寫入Offset offsetRanges.foreach { range writeOffset(connection, range) } connection.commit() } catch { case e: Exception connection.rollback() throw e } finally { connection.close() } } }這種方案的關(guān)鍵在于Offset的存儲位置必須和業(yè)務(wù)數(shù)據(jù)在同一個(gè)數(shù)據(jù)庫、同一個(gè)事務(wù)里。如果Offset存在ZooKeeper或Kafka里就無法和業(yè)務(wù)數(shù)據(jù)寫入構(gòu)成原子操作Exactly-Once就無從談起。需要提醒的是這種方案會顯著增加寫入延遲因?yàn)槊總€(gè)批次都要等事務(wù)提交完成。而且如果事務(wù)回滾整個(gè)批次的數(shù)據(jù)都要重寫對數(shù)據(jù)庫的壓力會成倍增加。在實(shí)際項(xiàng)目中我通常只在數(shù)據(jù)準(zhǔn)確性要求極高的場景比如金融交易記錄才用這種方案普通場景用冪等寫入就夠了。6. 性能調(diào)優(yōu)讓foreachRDD不再成為瓶頸6.1 批量寫入代替逐條寫入逐條寫入是foreachRDD最常見的性能殺手。每寫一條記錄就發(fā)一次網(wǎng)絡(luò)請求網(wǎng)絡(luò)往返時(shí)間RTT累積起來非?？捎^。假設(shè)RTT是1毫秒寫10000條記錄就是10秒這還沒算數(shù)據(jù)庫端的處理時(shí)間。批量寫入的思路是把多條記錄攢成一個(gè)批次一次性發(fā)給外部系統(tǒng)。對于JDBC可以用addBatch()和executeBatch()rdd.foreachPartition { partition val connection getConnection() val statement connection.prepareStatement(INSERT INTO ...) var count 0 partition.foreach { record statement.setString(1, record.field1) statement.setInt(2, record.field2) statement.addBatch() count 1 if (count % 500 0) { statement.executeBatch() count 0 } } if (count 0) { statement.executeBatch() } statement.close() connection.close() }批量大小需要根據(jù)記錄大小和數(shù)據(jù)庫配置來調(diào)整。一般來說500到1000條一批比較合適。太小了效果不明顯太大了可能超過數(shù)據(jù)庫的max_allowed_packet限制。6.2 控制分區(qū)數(shù)避免連接爆炸RDD的分區(qū)數(shù)直接決定了foreachPartition的并發(fā)度也決定了同時(shí)創(chuàng)建的連接數(shù)。如果分區(qū)數(shù)是500每個(gè)分區(qū)創(chuàng)建一個(gè)連接那么瞬間就會有500個(gè)連接打到數(shù)據(jù)庫上。大多數(shù)數(shù)據(jù)庫的默認(rèn)最大連接數(shù)是100到200直接就被打滿了?？刂品謪^(qū)數(shù)有幾種方式。如果數(shù)據(jù)源是Kafka可以用directStream的repartition或者調(diào)整spark.streaming.kafka.maxRatePerPartition來控制。如果是receive方式可以調(diào)整spark.streaming.blockInterval來間接影響分區(qū)數(shù)。更直接的方式是在foreachRDD之前調(diào)用rdd.repartition(n)或rdd.coalesce(n)來顯式控制分區(qū)數(shù)。但要注意repartition會觸發(fā)Shuffle有額外的開銷coalesce不觸發(fā)Shuffle但只能減少分區(qū)數(shù)不能增加。我的經(jīng)驗(yàn)是分區(qū)數(shù)控制在Executor總核數(shù)的2到3倍比較合適。比如你有10個(gè)Executor每個(gè)4核總共40核那么分區(qū)數(shù)設(shè)在80到120之間。這樣既能保證并發(fā)度又不會創(chuàng)建過多連接。6.3 異步寫入與背壓機(jī)制的取舍如果外部系統(tǒng)的寫入延遲比較高同步寫入會拖慢整個(gè)批次的處理速度進(jìn)而導(dǎo)致批次堆積。這時(shí)候可以考慮異步寫入——把數(shù)據(jù)放到一個(gè)隊(duì)列里由單獨(dú)的線程池負(fù)責(zé)寫入foreachRDD只負(fù)責(zé)入隊(duì)不等待寫入完成。object AsyncWriter { private val executor Executors.newFixedThreadPool(10) private val queue new LinkedBlockingQueue[Record](10000) def start(): Unit { (1 to 10).foreach { _ executor.submit(new Runnable { override def run(): Unit { while (true) { val record queue.take() // 寫入外部系統(tǒng) } } }) } } def write(record: Record): Unit { queue.put(record) // 隊(duì)列滿時(shí)會阻塞 } }異步寫入能顯著提升吞吐量但會帶來兩個(gè)問題。第一如果隊(duì)列滿了put操作會阻塞反而拖慢處理速度。第二異步寫入無法保證數(shù)據(jù)在批次完成時(shí)已經(jīng)落盤如果任務(wù)在這期間崩潰隊(duì)列里的數(shù)據(jù)就丟了。Spark Streaming本身有背壓機(jī)制spark.streaming.backpressure.enabledtrue可以根據(jù)處理速度動態(tài)調(diào)整數(shù)據(jù)攝入速率。但背壓只控制攝入不控制寫出。如果寫出是瓶頸背壓會導(dǎo)致攝入速率被壓低整體吞吐量上不去。這種情況下要么優(yōu)化寫出邏輯要么增加寫出并發(fā)度單純靠背壓解決不了問題。7. 那些文檔里不會寫的實(shí)戰(zhàn)經(jīng)驗(yàn)7.1 Checkpoint目錄的清理與版本兼容用了foreachRDD的任務(wù)通常都需要開啟Checkpoint因?yàn)橐４鍻ffset和DStream Graph的元信息。但Checkpoint目錄會隨著時(shí)間不斷增長如果不清理磁盤很快就會被占滿。Spark Streaming本身沒有提供自動清理Checkpoint的機(jī)制需要手動處理。常見的做法是寫一個(gè)定時(shí)任務(wù)定期刪除過期的Checkpoint文件。但要注意不能刪除正在使用的Checkpoint否則任務(wù)恢復(fù)時(shí)會失敗。另一個(gè)坑是Checkpoint的版本兼容性。如果你修改了DStream Graph的結(jié)構(gòu)比如增加了一個(gè)轉(zhuǎn)換操作從舊Checkpoint恢復(fù)時(shí)會報(bào)錯(cuò)因?yàn)樾屡fGraph不匹配。這種情況下只能刪除Checkpoint重新開始或者用spark.streaming.stopGracefullyOnShutdown優(yōu)雅停止后手動遷移。我的建議是如果業(yè)務(wù)邏輯可能頻繁變更盡量不要依賴Checkpoint恢復(fù)而是用外部存儲如ZooKeeper或數(shù)據(jù)庫來保存Offset每次啟動時(shí)從外部存儲讀取Offset重新構(gòu)建DStream。這樣DStream Graph的變更就不會受Checkpoint限制。7.2 foreachRDD內(nèi)部的異常處理策略foreachRDD內(nèi)部的異常如果沒被捕獲會導(dǎo)致整個(gè)批次失敗進(jìn)而觸發(fā)重試。如果異常是持續(xù)性的比如數(shù)據(jù)庫連不上重試會一直失敗任務(wù)會陷入死循環(huán)。合理的做法是在foreachRDD內(nèi)部做細(xì)粒度的異常處理。對于可恢復(fù)的異常如網(wǎng)絡(luò)抖動可以重試幾次對于不可恢復(fù)的異常如數(shù)據(jù)格式錯(cuò)誤應(yīng)該記錄日志并跳過而不是讓整個(gè)批次失敗。rdd.foreachPartition { partition partition.foreach { record try { // 寫入邏輯 } catch { case e: RetryableException // 重試邏輯 case e: Exception // 記錄到死信隊(duì)列或日志 logger.error(sFailed to process $record, e) } } }但要注意如果異常被吞掉了Spark Streaming會認(rèn)為批次處理成功Offset會正常提交。那些被跳過的數(shù)據(jù)就永久丟失了。所以跳過數(shù)據(jù)的前提是你已經(jīng)把失敗的數(shù)據(jù)記錄到了某個(gè)地方后續(xù)可以人工補(bǔ)償。7.3 本地測試與生產(chǎn)環(huán)境的差異本地測試foreachRDD時(shí)通常用local[*]模式所有Task在同一個(gè)JVM里跑連接池、單例對象的行為和生產(chǎn)環(huán)境完全不同。本地測試通過不代表生產(chǎn)環(huán)境沒問題。我踩過的一個(gè)坑是本地測試時(shí)連接池大小設(shè)成10跑得好好的到了生產(chǎn)環(huán)境每個(gè)Executor的JVM里都有一個(gè)連接池10個(gè)Executor就是100個(gè)連接直接把數(shù)據(jù)庫打滿了。本地測試時(shí)只有一個(gè)JVM連接池只有一份所以看不出問題。另一個(gè)差異是序列化。本地模式下閉包可能不需要真正序列化因?yàn)槎荚谕粋€(gè)JVM里所以一些序列化問題在本地不會暴露。到了集群模式閉包必須序列化問題才會顯現(xiàn)。所以foreachRDD相關(guān)的代碼一定要在真實(shí)集群環(huán)境上做集成測試至少要用yarn-client或standalone模式跑一遍才能發(fā)現(xiàn)連接池、序列化、并發(fā)這些方面的問題。7.4 監(jiān)控與告警別等任務(wù)掛了才發(fā)現(xiàn)foreachRDD的寫入失敗往往不會立即導(dǎo)致任務(wù)崩潰而是表現(xiàn)為批次處理時(shí)間逐漸變長、數(shù)據(jù)延遲越來越大。如果沒有監(jiān)控可能等到數(shù)據(jù)延遲幾小時(shí)才發(fā)現(xiàn)問題。需要監(jiān)控的指標(biāo)包括批次處理時(shí)間Batch Processing Time、調(diào)度延遲Scheduling Delay、每個(gè)批次的記錄數(shù)、寫入失敗率。Spark Streaming的Web UI提供了前兩個(gè)指標(biāo)后兩個(gè)需要自己在代碼里埋點(diǎn)。dstream.foreachRDD { rdd val startTime System.currentTimeMillis() val count rdd.count() rdd.foreachPartition { partition // 寫入邏輯 } val endTime System.currentTimeMillis() metricsReporter.report(write_latency, endTime - startTime) metricsReporter.report(record_count, count) }這些指標(biāo)可以推到Prometheus或Graphite配上告警規(guī)則。比如批次處理時(shí)間連續(xù)5分鐘超過batch interval的80%就觸發(fā)告警。這樣在問題惡化之前就能介入處理。8. 一個(gè)完整的生產(chǎn)級foreachRDD模板把前面講的要點(diǎn)整合起來下面是一個(gè)可以直接參考的生產(chǎn)級模板。這個(gè)模板包含了連接池管理、批量寫入、異常處理、冪等寫入和監(jiān)控埋點(diǎn)。object ProductionForeachRDDTemplate { def main(args: Array[String]): Unit { val conf new SparkConf() .setAppName(ProductionForeachRDD) .set(spark.streaming.backpressure.enabled, true) .set(spark.streaming.stopGracefullyOnShutdown, true) val ssc new StreamingContext(conf, Seconds(10)) ssc.checkpoint(hdfs://checkpoint/path) val kafkaParams Map( bootstrap.servers - kafka:9092, group.id - group1, auto.offset.reset - latest, enable.auto.commit - false ) val topics Set(topic1) val stream KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams) ) stream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges if (!rdd.isEmpty()) { val startTime System.currentTimeMillis() var totalCount 0L var failCount 0L rdd.foreachPartition { partition val pool ConnectionPoolManager.getPool() val connection pool.getConnection() connection.setAutoCommit(false) val statement connection.prepareStatement( INSERT INTO events (id, data, ts) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE data VALUES(data) ) var batchCount 0 try { partition.foreach { record try { val parsed parseRecord(record) statement.setString(1, parsed.id) statement.setString(2, parsed.data) statement.setLong(3, parsed.ts) statement.addBatch() batchCount 1 totalCount 1 if (batchCount % 500 0) { statement.executeBatch() connection.commit() batchCount 0 } } catch { case e: Exception failCount 1 logger.error(sFailed to process record: $record, e) } } if (batchCount 0) { statement.executeBatch() connection.commit() } } catch { case e: Exception connection.rollback() throw e } finally { statement.close() connection.close() } } // 提交Offset stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) val endTime System.currentTimeMillis() metricsReporter.report(batch_latency, endTime - startTime) metricsReporter.report(total_count, totalCount) metricsReporter.report(fail_count, failCount) } } ssc.start() ssc.awaitTermination() } }這個(gè)模板里有幾個(gè)設(shè)計(jì)決策值得說明。用createDirectStream而不是createStream是為了拿到OffsetRanges方便手動提交Offset。enable.auto.commit設(shè)為false避免Kafka自動提交Offset導(dǎo)致數(shù)據(jù)丟失。ON DUPLICATE KEY UPDATE保證了冪等性。批量大小設(shè)為500是經(jīng)驗(yàn)值可以根據(jù)實(shí)際情況調(diào)整。異常處理分兩層內(nèi)層捕獲單條記錄的異常避免一條壞數(shù)據(jù)導(dǎo)致整個(gè)批次失敗外層捕獲批量提交的異?；貪L事務(wù)并拋出讓Spark Streaming重試整個(gè)批次。連接池的配置需要根據(jù)Executor數(shù)量和數(shù)據(jù)庫承載能力來調(diào)整。如果每個(gè)Executor的maximumPoolSize是5有20個(gè)Executor那么總連接數(shù)是100。要確保數(shù)據(jù)庫的max_connections大于這個(gè)值并留出余量給其他應(yīng)用。監(jiān)控埋點(diǎn)建議推到外部監(jiān)控系統(tǒng)不要只打日志。日志在分布式環(huán)境下分散在各個(gè)Executor上排查問題很不方便。推到Prometheus或類似的系統(tǒng)可以統(tǒng)一查看和告警。這套模板我在多個(gè)生產(chǎn)任務(wù)里用過穩(wěn)定性不錯(cuò)。但每個(gè)項(xiàng)目的數(shù)據(jù)庫、數(shù)據(jù)量、延遲要求都不一樣具體參數(shù)需要根據(jù)實(shí)際情況調(diào)整。比如批量大小數(shù)據(jù)記錄大的時(shí)候要調(diào)小記錄小的時(shí)候可以調(diào)大。連接池大小也是數(shù)據(jù)庫性能好可以多開一些性能一般就要控制。最后說一個(gè)容易被忽略的點(diǎn)foreachRDD里的代碼在Driver上執(zhí)行的部分如果耗時(shí)太長比如做了復(fù)雜的計(jì)算會阻塞Driver的調(diào)度線程導(dǎo)致后續(xù)批次無法及時(shí)提交。所以foreachRDD方法體里盡量只做輕量的操作重活都放到rdd.foreachPartition里讓Executor去干。這個(gè)邊界意識是用好foreachRDD的關(guān)鍵。