優(yōu))
1. 為什么foreachRDD是Spark Streaming里最容易翻車的算子剛接觸Spark Streaming那會(huì)兒我對(duì)foreachRDD的理解就是把每個(gè)批次的數(shù)據(jù)拿出來處理一下看起來跟RDD的foreach差不多能有多難結(jié)果第一次在生產(chǎn)環(huán)境用它往外部存儲(chǔ)寫數(shù)據(jù)就踩了個(gè)大坑——每個(gè)批次都新建一個(gè)連接跑了幾分鐘直接把連接池打滿了任務(wù)全線卡死。后來復(fù)盤才發(fā)現(xiàn)問題根本不在業(yè)務(wù)邏輯而在于我對(duì)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ù)、但對(duì)foreachRDD的內(nèi)部機(jī)制和最佳實(shí)踐還不夠清晰的開發(fā)者。我會(huì)從執(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)度器會(huì)按照batch interval周期性地觸發(fā)作業(yè)。每個(gè)周期內(nèi)DStream Graph中所有的轉(zhuǎn)換操作會(huì)被編譯成一個(gè)Job交給底層的Spark引擎執(zhí)行。而foreachRDD就是這個(gè)Graph的終點(diǎn)站——它是一個(gè)Output Operation只有Output Operation才會(huì)真正觸發(fā)Job的提交。沒有Output Operation的DStream Graph就是一張廢紙什么都不會(huì)執(zhí)行。這一點(diǎn)跟RDD的惰性求值是一個(gè)道理。你寫了一堆map、filter、reduceByKey只要沒有actionSpark什么都不會(huì)做。DStream也一樣foreachRDD就是那個(gè)action級(jí)別的存在。區(qū)別在于RDD的action觸發(fā)一次計(jì)算就結(jié)束了而foreachRDD會(huì)在每個(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)部對(duì)RDD的操作就不一樣了。如果你在foreachRDD里面寫了rdd.foreach(...)那么rdd.foreach里面的代碼才會(huì)被分發(fā)到Executor上執(zhí)行。如果你寫的是rdd.collect()數(shù)據(jù)會(huì)被拉到Driver端。如果你寫的是rdd.saveAsTextFile(...)寫操作由Executor完成。這個(gè)邊界搞不清楚就會(huì)犯一個(gè)經(jīng)典錯(cuò)誤在foreachRDD的方法體里直接創(chuàng)建數(shù)據(jù)庫連接。因?yàn)榉椒w在Driver上執(zhí)行你創(chuàng)建的連接對(duì)象在Driver上然后你試圖在rdd.foreach里用它——這個(gè)連接對(duì)象需要被序列化后發(fā)送到Executor而數(shù)據(jù)庫連接根本不可序列化直接報(bào)NotSerializableException。就算某些連接碰巧能序列化每個(gè)Executor上拿到的也是一個(gè)反序列化后的副本跟Driver上的原始連接沒有任何關(guān)系用完了也不會(huì)關(guān)閉連接泄漏是必然的。2.3 批次間隔與foreachRDD調(diào)用頻率的關(guān)系foreachRDD的調(diào)用頻率等于batch interval的倒數(shù)。如果你設(shè)置的batch interval是5秒那么foreachRDD的方法體每5秒被執(zhí)行一次里面的RDD操作也會(huì)每5秒觸發(fā)一次。這個(gè)頻率直接決定了你對(duì)外部系統(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)對(duì)連接創(chuàng)建比較敏感要么把batch interval調(diào)大要么在foreachRDD內(nèi)部做連接池復(fù)用。這兩個(gè)方向后面都會(huì)詳細(xì)講。3. 連接管理的三種寫法與它們的性能差距3.1 最直覺但最致命的寫法每個(gè)RDD建一次連接先看一段很多人一開始都會(huì)寫的代碼dstream.foreachRDD { rdd val connection createConnection() // 在Driver上執(zhí)行 rdd.foreach { record connection.write(record) // 連接被序列化到Executor必然失敗 } connection.close() }這段代碼有兩個(gè)致命問題。第一createConnection()在Driver上執(zhí)行得到的連接對(duì)象要隨rdd.foreach的閉包一起序列化發(fā)送到Executor絕大多數(shù)數(shù)據(jù)庫連接不支持序列化直接拋異常。第二即使某個(gè)連接庫支持序列化每個(gè)Executor上反序列化出來的連接是獨(dú)立的副本connection.close()在Driver上調(diào)用關(guān)的是Driver上的那個(gè)連接Executor上的連接永遠(yuǎn)不會(huì)被關(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ù)量級(jí)的。但即便如此如果分區(qū)數(shù)很多每個(gè)批次仍然會(huì)創(chuàng)建大量連接。假設(shè)你的RDD有200個(gè)分區(qū)batch interval是10秒那么每10秒就會(huì)創(chuàng)建和銷毀200個(gè)連接一分鐘就是1200個(gè)。對(duì)于連接創(chuàng)建開銷大的數(shù)據(jù)庫來說這個(gè)壓力依然不小。3.2 連接池方案用單例對(duì)象管理連接生命周期連接池的核心思路是讓連接在Executor的JVM生命周期內(nèi)復(fù)用而不是每個(gè)批次都重新創(chuàng)建。實(shí)現(xiàn)方式是利用Scala的object單例特性——每個(gè)Executor的JVM里這個(gè)object只會(huì)被初始化一次里面的連接池也就只創(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è)對(duì)象本身會(huì)被序列化到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è)太大反而會(huì)浪費(fèi)資源還可能把數(shù)據(jù)庫的連接數(shù)打滿。3.3 三種方案的量化對(duì)比方案連接創(chuàng)建頻率每批次連接數(shù)適用場(chǎng)景風(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)的場(chǎng)景。4. 序列化問題為什么你的連接對(duì)象總是報(bào)錯(cuò)4.1 閉包序列化的底層機(jī)制Spark在分發(fā)Task到Executor時(shí)會(huì)把Task對(duì)應(yīng)的閉包序列化成字節(jié)數(shù)組。閉包會(huì)捕獲它引用的外部變量這些變量必須支持序列化。foreachRDD的方法體在Driver上執(zhí)行但方法體內(nèi)部的rdd.foreach或rdd.foreachPartition的閉包會(huì)被序列化發(fā)送到Executor。如果你在foreachRDD的方法體里定義了一個(gè)連接對(duì)象然后在rdd.foreach里引用它這個(gè)連接對(duì)象就會(huì)被閉包捕獲進(jìn)而被序列化。數(shù)據(jù)庫連接、網(wǎng)絡(luò)連接、文件句柄這些資源都不支持序列化所以會(huì)直接報(bào)java.io.NotSerializableException。更隱蔽的情況是你定義了一個(gè)包含連接字段的類這個(gè)類實(shí)現(xiàn)了Serializable接口但連接字段本身不可序列化。這種情況下序列化會(huì)在運(yùn)行時(shí)失敗而且報(bào)錯(cuò)信息可能不會(huì)直接指向連接字段排查起來比較費(fèi)勁。4.2 transient注解的正確使用姿勢(shì)transient是Scala和Java都支持的注解作用是標(biāo)記某個(gè)字段不參與序列化。在Spark Streaming的場(chǎng)景里它通常用來標(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í)會(huì)被跳過。到了Executor端反序列化后connection是null需要在第一次使用時(shí)重新創(chuàng)建。這種惰性初始化的模式在Spark里非常常見。但要注意transient只是讓字段不被序列化它不會(huì)自動(dòng)幫你重新初始化。你必須在代碼里顯式判斷null并創(chuàng)建連接否則會(huì)報(bào)NullPointerException。而且如果多個(gè)Task共享同一個(gè)對(duì)象實(shí)例還需要考慮線程安全問題。4.3 一個(gè)真實(shí)的序列化報(bào)錯(cuò)排查過程有一次同事找我?guī)兔匆粋€(gè)任務(wù)報(bào)錯(cuò)信息是Task not serializable但代碼里看起來沒有任何明顯的不可序列化對(duì)象。我讓他把foreachRDD里的代碼逐行注釋最后定位到一行看起來人畜無害的代碼val logger LoggerFactory.getLogger(this.getClass) dstream.foreachRDD { rdd rdd.foreach { record logger.info(sProcessing $record) // 這里引用了logger } }問題出在logger上。LoggerFactory.getLogger返回的Logger對(duì)象內(nèi)部持有大量不可序列化的狀態(tài)被閉包捕獲后導(dǎo)致序列化失敗。解決辦法是在rdd.foreach內(nèi)部重新獲取Logger或者用transient標(biāo)記Logger字段。這個(gè)案例說明序列化問題不一定來自你顯式創(chuàng)建的連接對(duì)象任何被閉包捕獲的、不可序列化的對(duì)象都可能成為罪魁禍?zhǔn)住Ｅ挪榈臅r(shí)候最有效的方法就是逐行注釋二分定位。5. 冪等性與Exactly-OnceforeachRDD繞不開的難題5.1 At-Least-Once語義下重復(fù)寫入的根源Spark Streaming默認(rèn)提供的是At-Least-Once語義。這意味著在發(fā)生故障恢復(fù)時(shí)某些批次的數(shù)據(jù)可能會(huì)被重復(fù)處理。對(duì)于foreachRDD來說重復(fù)處理就意味著重復(fù)寫入外部系統(tǒng)。重復(fù)寫入的根源在于Spark Streaming的Checkpoint機(jī)制和WALWrite-Ahead Log。當(dāng)Executor失敗時(shí)Driver會(huì)根據(jù)Checkpoint信息重新調(diào)度失敗的批次。如果失敗發(fā)生在數(shù)據(jù)已經(jīng)寫入外部系統(tǒng)之后、但Checkpoint還沒更新之前那么這個(gè)批次就會(huì)被重新執(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è)批次會(huì)被重新處理同樣的數(shù)據(jù)會(huì)被再次寫入MySQL。如果你的寫入操作是簡單的INSERTMySQL里就會(huì)出現(xiàn)重復(fù)記錄。5.2 用冪等寫入對(duì)抗重復(fù)數(shù)據(jù)解決重復(fù)寫入最直接的辦法是讓寫入操作本身具備冪等性。冪等的意思是同樣的操作執(zhí)行一次和執(zhí)行多次對(duì)外部系統(tǒng)的影響是一樣的。對(duì)于MySQL常見的冪等方案有三種。第一種是用INSERT ... ON DUPLICATE KEY UPDATE前提是表上有唯一鍵。第二種是先用REPLACE INTO效果類似。第三種是先DELETE再INSERT但這種方式在并發(fā)場(chǎng)景下可能有問題。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)對(duì)于HBase可以利用RowKey的唯一性用Put操作天然實(shí)現(xiàn)冪等——同樣的RowKey重復(fù)寫入后寫的會(huì)覆蓋先寫的結(jié)果是一樣的。對(duì)于Kafka可以啟用Producer的冪等性enable.idempotencetrueKafka會(huì)為每條消息分配序列號(hào)Broker端會(huì)去重。但要注意Kafka的冪等性只在單個(gè)Producer會(huì)話內(nèi)有效跨會(huì)話的重復(fù)還是需要業(yè)務(wù)層處理。5.3 事務(wù)性寫入與手動(dòng)提交Offset的配合如果你需要更強(qiáng)的Exactly-Once保證就需要把數(shù)據(jù)寫入和Offset提交放在同一個(gè)事務(wù)里?；舅悸肥窃谔幚砻總€(gè)批次時(shí)先開啟事務(wù)寫入數(shù)據(jù)同時(shí)把Offset信息也寫入同一個(gè)事務(wù)最后提交事務(wù)。如果事務(wù)失敗Offset不會(huì)提交批次會(huì)被重新處理。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的存儲(chǔ)位置必須和業(yè)務(wù)數(shù)據(jù)在同一個(gè)數(shù)據(jù)庫、同一個(gè)事務(wù)里。如果Offset存在ZooKeeper或Kafka里就無法和業(yè)務(wù)數(shù)據(jù)寫入構(gòu)成原子操作Exactly-Once就無從談起。需要提醒的是這種方案會(huì)顯著增加寫入延遲因?yàn)槊總€(gè)批次都要等事務(wù)提交完成。而且如果事務(wù)回滾整個(gè)批次的數(shù)據(jù)都要重寫對(duì)數(shù)據(jù)庫的壓力會(huì)成倍增加。在實(shí)際項(xiàng)目中我通常只在數(shù)據(jù)準(zhǔn)確性要求極高的場(chǎng)景比如金融交易記錄才用這種方案普通場(chǎng)景用冪等寫入就夠了。6. 性能調(diào)優(yōu)讓foreachRDD不再成為瓶頸6.1 批量寫入代替逐條寫入逐條寫入是foreachRDD最常見的性能殺手。每寫一條記錄就發(fā)一次網(wǎng)絡(luò)請(qǐng)求網(wǎng)絡(luò)往返時(shí)間RTT累積起來非?？捎^。假設(shè)RTT是1毫秒寫10000條記錄就是10秒這還沒算數(shù)據(jù)庫端的處理時(shí)間。批量寫入的思路是把多條記錄攢成一個(gè)批次一次性發(fā)給外部系統(tǒng)。對(duì)于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è)連接那么瞬間就會(huì)有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會(huì)觸發(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ā)度又不會(huì)創(chuàng)建過多連接。6.3 異步寫入與背壓機(jī)制的取舍如果外部系統(tǒng)的寫入延遲比較高同步寫入會(huì)拖慢整個(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í)會(huì)阻塞 } }異步寫入能顯著提升吞吐量但會(huì)帶來兩個(gè)問題。第一如果隊(duì)列滿了put操作會(huì)阻塞反而拖慢處理速度。第二異步寫入無法保證數(shù)據(jù)在批次完成時(shí)已經(jīng)落盤如果任務(wù)在這期間崩潰隊(duì)列里的數(shù)據(jù)就丟了。Spark Streaming本身有背壓機(jī)制spark.streaming.backpressure.enabledtrue可以根據(jù)處理速度動(dòng)態(tài)調(diào)整數(shù)據(jù)攝入速率。但背壓只控制攝入不控制寫出。如果寫出是瓶頸背壓會(huì)導(dǎo)致攝入速率被壓低整體吞吐量上不去。這種情況下要么優(yōu)化寫出邏輯要么增加寫出并發(fā)度單純靠背壓解決不了問題。7. 那些文檔里不會(huì)寫的實(shí)戰(zhàn)經(jīng)驗(yàn)7.1 Checkpoint目錄的清理與版本兼容用了foreachRDD的任務(wù)通常都需要開啟Checkpoint因?yàn)橐４鍻ffset和DStream Graph的元信息。但Checkpoint目錄會(huì)隨著時(shí)間不斷增長如果不清理磁盤很快就會(huì)被占滿。Spark Streaming本身沒有提供自動(dòng)清理Checkpoint的機(jī)制需要手動(dòng)處理。常見的做法是寫一個(gè)定時(shí)任務(wù)定期刪除過期的Checkpoint文件。但要注意不能刪除正在使用的Checkpoint否則任務(wù)恢復(fù)時(shí)會(huì)失敗。另一個(gè)坑是Checkpoint的版本兼容性。如果你修改了DStream Graph的結(jié)構(gòu)比如增加了一個(gè)轉(zhuǎn)換操作從舊Checkpoint恢復(fù)時(shí)會(huì)報(bào)錯(cuò)因?yàn)樾屡fGraph不匹配。這種情況下只能刪除Checkpoint重新開始或者用spark.streaming.stopGracefullyOnShutdown優(yōu)雅停止后手動(dòng)遷移。我的建議是如果業(yè)務(wù)邏輯可能頻繁變更盡量不要依賴Checkpoint恢復(fù)而是用外部存儲(chǔ)如ZooKeeper或數(shù)據(jù)庫來保存Offset每次啟動(dòng)時(shí)從外部存儲(chǔ)讀取Offset重新構(gòu)建DStream。這樣DStream Graph的變更就不會(huì)受Checkpoint限制。7.2 foreachRDD內(nèi)部的異常處理策略foreachRDD內(nèi)部的異常如果沒被捕獲會(huì)導(dǎo)致整個(gè)批次失敗進(jìn)而觸發(fā)重試。如果異常是持續(xù)性的比如數(shù)據(jù)庫連不上重試會(huì)一直失敗任務(wù)會(huì)陷入死循環(huán)。合理的做法是在foreachRDD內(nèi)部做細(xì)粒度的異常處理。對(duì)于可恢復(fù)的異常如網(wǎng)絡(luò)抖動(dòng)可以重試幾次對(duì)于不可恢復(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會(huì)認(rèn)為批次處理成功Offset會(huì)正常提交。那些被跳過的數(shù)據(jù)就永久丟失了。所以跳過數(shù)據(jù)的前提是你已經(jīng)把失敗的數(shù)據(jù)記錄到了某個(gè)地方后續(xù)可以人工補(bǔ)償。7.3 本地測(cè)試與生產(chǎn)環(huán)境的差異本地測(cè)試foreachRDD時(shí)通常用local[*]模式所有Task在同一個(gè)JVM里跑連接池、單例對(duì)象的行為和生產(chǎn)環(huán)境完全不同。本地測(cè)試通過不代表生產(chǎn)環(huán)境沒問題。我踩過的一個(gè)坑是本地測(cè)試時(shí)連接池大小設(shè)成10跑得好好的到了生產(chǎn)環(huán)境每個(gè)Executor的JVM里都有一個(gè)連接池10個(gè)Executor就是100個(gè)連接直接把數(shù)據(jù)庫打滿了。本地測(cè)試時(shí)只有一個(gè)JVM連接池只有一份所以看不出問題。另一個(gè)差異是序列化。本地模式下閉包可能不需要真正序列化因?yàn)槎荚谕粋€(gè)JVM里所以一些序列化問題在本地不會(huì)暴露。到了集群模式閉包必須序列化問題才會(huì)顯現(xiàn)。所以foreachRDD相關(guān)的代碼一定要在真實(shí)集群環(huán)境上做集成測(cè)試至少要用yarn-client或standalone模式跑一遍才能發(fā)現(xiàn)連接池、序列化、并發(fā)這些方面的問題。7.4 監(jiān)控與告警別等任務(wù)掛了才發(fā)現(xiàn)foreachRDD的寫入失敗往往不會(huì)立即導(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)級(jí)foreachRDD模板把前面講的要點(diǎn)整合起來下面是一個(gè)可以直接參考的生產(chǎn)級(jí)模板。這個(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方便手動(dòng)提交Offset。enable.auto.commit設(shè)為false避免Kafka自動(dòng)提交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ì)算會(huì)阻塞Driver的調(diào)度線程導(dǎo)致后續(xù)批次無法及時(shí)提交。所以foreachRDD方法體里盡量只做輕量的操作重活都放到rdd.foreachPartition里讓Executor去干。這個(gè)邊界意識(shí)是用好foreachRDD的關(guān)鍵。 Spark Streaming中foreachRDD的避坑指南：連接管理與性能調(diào)優(yōu)
1. 為什么foreachRDD是Spark Streaming里最容易翻車的算子剛接觸Spark Streaming那會(huì)兒我對(duì)foreachRDD的理解就是把每個(gè)批次的數(shù)據(jù)拿出來處理一下看起來跟RDD的foreach差不多能有多難結(jié)果第一次在生產(chǎn)環(huán)境用它往外部存儲(chǔ)寫數(shù)據(jù)就踩了個(gè)大坑——每個(gè)批次都新建一個(gè)連接跑了幾分鐘直接把連接池打滿了任務(wù)全線卡死。后來復(fù)盤才發(fā)現(xiàn)問題根本不在業(yè)務(wù)邏輯而在于我對(duì)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ù)、但對(duì)foreachRDD的內(nèi)部機(jī)制和最佳實(shí)踐還不夠清晰的開發(fā)者。我會(huì)從執(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)度器會(huì)按照batch interval周期性地觸發(fā)作業(yè)。每個(gè)周期內(nèi)DStream Graph中所有的轉(zhuǎn)換操作會(huì)被編譯成一個(gè)Job交給底層的Spark引擎執(zhí)行。而foreachRDD就是這個(gè)Graph的終點(diǎn)站——它是一個(gè)Output Operation只有Output Operation才會(huì)真正觸發(fā)Job的提交。沒有Output Operation的DStream Graph就是一張廢紙什么都不會(huì)執(zhí)行。這一點(diǎn)跟RDD的惰性求值是一個(gè)道理。你寫了一堆map、filter、reduceByKey只要沒有actionSpark什么都不會(huì)做。DStream也一樣foreachRDD就是那個(gè)action級(jí)別的存在。區(qū)別在于RDD的action觸發(fā)一次計(jì)算就結(jié)束了而foreachRDD會(huì)在每個(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)部對(duì)RDD的操作就不一樣了。如果你在foreachRDD里面寫了rdd.foreach(...)那么rdd.foreach里面的代碼才會(huì)被分發(fā)到Executor上執(zhí)行。如果你寫的是rdd.collect()數(shù)據(jù)會(huì)被拉到Driver端。如果你寫的是rdd.saveAsTextFile(...)寫操作由Executor完成。這個(gè)邊界搞不清楚就會(huì)犯一個(gè)經(jīng)典錯(cuò)誤在foreachRDD的方法體里直接創(chuàng)建數(shù)據(jù)庫連接。因?yàn)榉椒w在Driver上執(zhí)行你創(chuàng)建的連接對(duì)象在Driver上然后你試圖在rdd.foreach里用它——這個(gè)連接對(duì)象需要被序列化后發(fā)送到Executor而數(shù)據(jù)庫連接根本不可序列化直接報(bào)NotSerializableException。就算某些連接碰巧能序列化每個(gè)Executor上拿到的也是一個(gè)反序列化后的副本跟Driver上的原始連接沒有任何關(guān)系用完了也不會(huì)關(guān)閉連接泄漏是必然的。2.3 批次間隔與foreachRDD調(diào)用頻率的關(guān)系foreachRDD的調(diào)用頻率等于batch interval的倒數(shù)。如果你設(shè)置的batch interval是5秒那么foreachRDD的方法體每5秒被執(zhí)行一次里面的RDD操作也會(huì)每5秒觸發(fā)一次。這個(gè)頻率直接決定了你對(duì)外部系統(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)對(duì)連接創(chuàng)建比較敏感要么把batch interval調(diào)大要么在foreachRDD內(nèi)部做連接池復(fù)用。這兩個(gè)方向后面都會(huì)詳細(xì)講。3. 連接管理的三種寫法與它們的性能差距3.1 最直覺但最致命的寫法每個(gè)RDD建一次連接先看一段很多人一開始都會(huì)寫的代碼dstream.foreachRDD { rdd val connection createConnection() // 在Driver上執(zhí)行 rdd.foreach { record connection.write(record) // 連接被序列化到Executor必然失敗 } connection.close() }這段代碼有兩個(gè)致命問題。第一createConnection()在Driver上執(zhí)行得到的連接對(duì)象要隨rdd.foreach的閉包一起序列化發(fā)送到Executor絕大多數(shù)數(shù)據(jù)庫連接不支持序列化直接拋異常。第二即使某個(gè)連接庫支持序列化每個(gè)Executor上反序列化出來的連接是獨(dú)立的副本connection.close()在Driver上調(diào)用關(guān)的是Driver上的那個(gè)連接Executor上的連接永遠(yuǎn)不會(huì)被關(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ù)量級(jí)的。但即便如此如果分區(qū)數(shù)很多每個(gè)批次仍然會(huì)創(chuàng)建大量連接。假設(shè)你的RDD有200個(gè)分區(qū)batch interval是10秒那么每10秒就會(huì)創(chuàng)建和銷毀200個(gè)連接一分鐘就是1200個(gè)。對(duì)于連接創(chuàng)建開銷大的數(shù)據(jù)庫來說這個(gè)壓力依然不小。3.2 連接池方案用單例對(duì)象管理連接生命周期連接池的核心思路是讓連接在Executor的JVM生命周期內(nèi)復(fù)用而不是每個(gè)批次都重新創(chuàng)建。實(shí)現(xiàn)方式是利用Scala的object單例特性——每個(gè)Executor的JVM里這個(gè)object只會(huì)被初始化一次里面的連接池也就只創(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è)對(duì)象本身會(huì)被序列化到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è)太大反而會(huì)浪費(fèi)資源還可能把數(shù)據(jù)庫的連接數(shù)打滿。3.3 三種方案的量化對(duì)比方案連接創(chuàng)建頻率每批次連接數(shù)適用場(chǎng)景風(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)的場(chǎng)景。4. 序列化問題為什么你的連接對(duì)象總是報(bào)錯(cuò)4.1 閉包序列化的底層機(jī)制Spark在分發(fā)Task到Executor時(shí)會(huì)把Task對(duì)應(yīng)的閉包序列化成字節(jié)數(shù)組。閉包會(huì)捕獲它引用的外部變量這些變量必須支持序列化。foreachRDD的方法體在Driver上執(zhí)行但方法體內(nèi)部的rdd.foreach或rdd.foreachPartition的閉包會(huì)被序列化發(fā)送到Executor。如果你在foreachRDD的方法體里定義了一個(gè)連接對(duì)象然后在rdd.foreach里引用它這個(gè)連接對(duì)象就會(huì)被閉包捕獲進(jìn)而被序列化。數(shù)據(jù)庫連接、網(wǎng)絡(luò)連接、文件句柄這些資源都不支持序列化所以會(huì)直接報(bào)java.io.NotSerializableException。更隱蔽的情況是你定義了一個(gè)包含連接字段的類這個(gè)類實(shí)現(xiàn)了Serializable接口但連接字段本身不可序列化。這種情況下序列化會(huì)在運(yùn)行時(shí)失敗而且報(bào)錯(cuò)信息可能不會(huì)直接指向連接字段排查起來比較費(fèi)勁。4.2 transient注解的正確使用姿勢(shì)transient是Scala和Java都支持的注解作用是標(biāo)記某個(gè)字段不參與序列化。在Spark Streaming的場(chǎng)景里它通常用來標(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í)會(huì)被跳過。到了Executor端反序列化后connection是null需要在第一次使用時(shí)重新創(chuàng)建。這種惰性初始化的模式在Spark里非常常見。但要注意transient只是讓字段不被序列化它不會(huì)自動(dòng)幫你重新初始化。你必須在代碼里顯式判斷null并創(chuàng)建連接否則會(huì)報(bào)NullPointerException。而且如果多個(gè)Task共享同一個(gè)對(duì)象實(shí)例還需要考慮線程安全問題。4.3 一個(gè)真實(shí)的序列化報(bào)錯(cuò)排查過程有一次同事找我?guī)兔匆粋€(gè)任務(wù)報(bào)錯(cuò)信息是Task not serializable但代碼里看起來沒有任何明顯的不可序列化對(duì)象。我讓他把foreachRDD里的代碼逐行注釋最后定位到一行看起來人畜無害的代碼val logger LoggerFactory.getLogger(this.getClass) dstream.foreachRDD { rdd rdd.foreach { record logger.info(sProcessing $record) // 這里引用了logger } }問題出在logger上。LoggerFactory.getLogger返回的Logger對(duì)象內(nèi)部持有大量不可序列化的狀態(tài)被閉包捕獲后導(dǎo)致序列化失敗。解決辦法是在rdd.foreach內(nèi)部重新獲取Logger或者用transient標(biāo)記Logger字段。這個(gè)案例說明序列化問題不一定來自你顯式創(chuàng)建的連接對(duì)象任何被閉包捕獲的、不可序列化的對(duì)象都可能成為罪魁禍?zhǔn)住Ｅ挪榈臅r(shí)候最有效的方法就是逐行注釋二分定位。5. 冪等性與Exactly-OnceforeachRDD繞不開的難題5.1 At-Least-Once語義下重復(fù)寫入的根源Spark Streaming默認(rèn)提供的是At-Least-Once語義。這意味著在發(fā)生故障恢復(fù)時(shí)某些批次的數(shù)據(jù)可能會(huì)被重復(fù)處理。對(duì)于foreachRDD來說重復(fù)處理就意味著重復(fù)寫入外部系統(tǒng)。重復(fù)寫入的根源在于Spark Streaming的Checkpoint機(jī)制和WALWrite-Ahead Log。當(dāng)Executor失敗時(shí)Driver會(huì)根據(jù)Checkpoint信息重新調(diào)度失敗的批次。如果失敗發(fā)生在數(shù)據(jù)已經(jīng)寫入外部系統(tǒng)之后、但Checkpoint還沒更新之前那么這個(gè)批次就會(huì)被重新執(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è)批次會(huì)被重新處理同樣的數(shù)據(jù)會(huì)被再次寫入MySQL。如果你的寫入操作是簡單的INSERTMySQL里就會(huì)出現(xiàn)重復(fù)記錄。5.2 用冪等寫入對(duì)抗重復(fù)數(shù)據(jù)解決重復(fù)寫入最直接的辦法是讓寫入操作本身具備冪等性。冪等的意思是同樣的操作執(zhí)行一次和執(zhí)行多次對(duì)外部系統(tǒng)的影響是一樣的。對(duì)于MySQL常見的冪等方案有三種。第一種是用INSERT ... ON DUPLICATE KEY UPDATE前提是表上有唯一鍵。第二種是先用REPLACE INTO效果類似。第三種是先DELETE再INSERT但這種方式在并發(fā)場(chǎng)景下可能有問題。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)對(duì)于HBase可以利用RowKey的唯一性用Put操作天然實(shí)現(xiàn)冪等——同樣的RowKey重復(fù)寫入后寫的會(huì)覆蓋先寫的結(jié)果是一樣的。對(duì)于Kafka可以啟用Producer的冪等性enable.idempotencetrueKafka會(huì)為每條消息分配序列號(hào)Broker端會(huì)去重。但要注意Kafka的冪等性只在單個(gè)Producer會(huì)話內(nèi)有效跨會(huì)話的重復(fù)還是需要業(yè)務(wù)層處理。5.3 事務(wù)性寫入與手動(dòng)提交Offset的配合如果你需要更強(qiáng)的Exactly-Once保證就需要把數(shù)據(jù)寫入和Offset提交放在同一個(gè)事務(wù)里?；舅悸肥窃谔幚砻總€(gè)批次時(shí)先開啟事務(wù)寫入數(shù)據(jù)同時(shí)把Offset信息也寫入同一個(gè)事務(wù)最后提交事務(wù)。如果事務(wù)失敗Offset不會(huì)提交批次會(huì)被重新處理。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的存儲(chǔ)位置必須和業(yè)務(wù)數(shù)據(jù)在同一個(gè)數(shù)據(jù)庫、同一個(gè)事務(wù)里。如果Offset存在ZooKeeper或Kafka里就無法和業(yè)務(wù)數(shù)據(jù)寫入構(gòu)成原子操作Exactly-Once就無從談起。需要提醒的是這種方案會(huì)顯著增加寫入延遲因?yàn)槊總€(gè)批次都要等事務(wù)提交完成。而且如果事務(wù)回滾整個(gè)批次的數(shù)據(jù)都要重寫對(duì)數(shù)據(jù)庫的壓力會(huì)成倍增加。在實(shí)際項(xiàng)目中我通常只在數(shù)據(jù)準(zhǔn)確性要求極高的場(chǎng)景比如金融交易記錄才用這種方案普通場(chǎng)景用冪等寫入就夠了。6. 性能調(diào)優(yōu)讓foreachRDD不再成為瓶頸6.1 批量寫入代替逐條寫入逐條寫入是foreachRDD最常見的性能殺手。每寫一條記錄就發(fā)一次網(wǎng)絡(luò)請(qǐng)求網(wǎng)絡(luò)往返時(shí)間RTT累積起來非?？捎^。假設(shè)RTT是1毫秒寫10000條記錄就是10秒這還沒算數(shù)據(jù)庫端的處理時(shí)間。批量寫入的思路是把多條記錄攢成一個(gè)批次一次性發(fā)給外部系統(tǒng)。對(duì)于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è)連接那么瞬間就會(huì)有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會(huì)觸發(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ā)度又不會(huì)創(chuàng)建過多連接。6.3 異步寫入與背壓機(jī)制的取舍如果外部系統(tǒng)的寫入延遲比較高同步寫入會(huì)拖慢整個(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í)會(huì)阻塞 } }異步寫入能顯著提升吞吐量但會(huì)帶來兩個(gè)問題。第一如果隊(duì)列滿了put操作會(huì)阻塞反而拖慢處理速度。第二異步寫入無法保證數(shù)據(jù)在批次完成時(shí)已經(jīng)落盤如果任務(wù)在這期間崩潰隊(duì)列里的數(shù)據(jù)就丟了。Spark Streaming本身有背壓機(jī)制spark.streaming.backpressure.enabledtrue可以根據(jù)處理速度動(dòng)態(tài)調(diào)整數(shù)據(jù)攝入速率。但背壓只控制攝入不控制寫出。如果寫出是瓶頸背壓會(huì)導(dǎo)致攝入速率被壓低整體吞吐量上不去。這種情況下要么優(yōu)化寫出邏輯要么增加寫出并發(fā)度單純靠背壓解決不了問題。7. 那些文檔里不會(huì)寫的實(shí)戰(zhàn)經(jīng)驗(yàn)7.1 Checkpoint目錄的清理與版本兼容用了foreachRDD的任務(wù)通常都需要開啟Checkpoint因?yàn)橐４鍻ffset和DStream Graph的元信息。但Checkpoint目錄會(huì)隨著時(shí)間不斷增長如果不清理磁盤很快就會(huì)被占滿。Spark Streaming本身沒有提供自動(dòng)清理Checkpoint的機(jī)制需要手動(dòng)處理。常見的做法是寫一個(gè)定時(shí)任務(wù)定期刪除過期的Checkpoint文件。但要注意不能刪除正在使用的Checkpoint否則任務(wù)恢復(fù)時(shí)會(huì)失敗。另一個(gè)坑是Checkpoint的版本兼容性。如果你修改了DStream Graph的結(jié)構(gòu)比如增加了一個(gè)轉(zhuǎn)換操作從舊Checkpoint恢復(fù)時(shí)會(huì)報(bào)錯(cuò)因?yàn)樾屡fGraph不匹配。這種情況下只能刪除Checkpoint重新開始或者用spark.streaming.stopGracefullyOnShutdown優(yōu)雅停止后手動(dòng)遷移。我的建議是如果業(yè)務(wù)邏輯可能頻繁變更盡量不要依賴Checkpoint恢復(fù)而是用外部存儲(chǔ)如ZooKeeper或數(shù)據(jù)庫來保存Offset每次啟動(dòng)時(shí)從外部存儲(chǔ)讀取Offset重新構(gòu)建DStream。這樣DStream Graph的變更就不會(huì)受Checkpoint限制。7.2 foreachRDD內(nèi)部的異常處理策略foreachRDD內(nèi)部的異常如果沒被捕獲會(huì)導(dǎo)致整個(gè)批次失敗進(jìn)而觸發(fā)重試。如果異常是持續(xù)性的比如數(shù)據(jù)庫連不上重試會(huì)一直失敗任務(wù)會(huì)陷入死循環(huán)。合理的做法是在foreachRDD內(nèi)部做細(xì)粒度的異常處理。對(duì)于可恢復(fù)的異常如網(wǎng)絡(luò)抖動(dòng)可以重試幾次對(duì)于不可恢復(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會(huì)認(rèn)為批次處理成功Offset會(huì)正常提交。那些被跳過的數(shù)據(jù)就永久丟失了。所以跳過數(shù)據(jù)的前提是你已經(jīng)把失敗的數(shù)據(jù)記錄到了某個(gè)地方后續(xù)可以人工補(bǔ)償。7.3 本地測(cè)試與生產(chǎn)環(huán)境的差異本地測(cè)試foreachRDD時(shí)通常用local[*]模式所有Task在同一個(gè)JVM里跑連接池、單例對(duì)象的行為和生產(chǎn)環(huán)境完全不同。本地測(cè)試通過不代表生產(chǎn)環(huán)境沒問題。我踩過的一個(gè)坑是本地測(cè)試時(shí)連接池大小設(shè)成10跑得好好的到了生產(chǎn)環(huán)境每個(gè)Executor的JVM里都有一個(gè)連接池10個(gè)Executor就是100個(gè)連接直接把數(shù)據(jù)庫打滿了。本地測(cè)試時(shí)只有一個(gè)JVM連接池只有一份所以看不出問題。另一個(gè)差異是序列化。本地模式下閉包可能不需要真正序列化因?yàn)槎荚谕粋€(gè)JVM里所以一些序列化問題在本地不會(huì)暴露。到了集群模式閉包必須序列化問題才會(huì)顯現(xiàn)。所以foreachRDD相關(guān)的代碼一定要在真實(shí)集群環(huán)境上做集成測(cè)試至少要用yarn-client或standalone模式跑一遍才能發(fā)現(xiàn)連接池、序列化、并發(fā)這些方面的問題。7.4 監(jiān)控與告警別等任務(wù)掛了才發(fā)現(xiàn)foreachRDD的寫入失敗往往不會(huì)立即導(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)級(jí)foreachRDD模板把前面講的要點(diǎn)整合起來下面是一個(gè)可以直接參考的生產(chǎn)級(jí)模板。這個(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方便手動(dòng)提交Offset。enable.auto.commit設(shè)為false避免Kafka自動(dòng)提交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ì)算會(huì)阻塞Driver的調(diào)度線程導(dǎo)致后續(xù)批次無法及時(shí)提交。所以foreachRDD方法體里盡量只做輕量的操作重活都放到rdd.foreachPartition里讓Executor去干。這個(gè)邊界意識(shí)是用好foreachRDD的關(guān)鍵。