1. 項(xiàng)目概述為什么Spark Streaming依然是實(shí)時(shí)計(jì)算的基石最近和幾個(gè)做數(shù)據(jù)平臺(tái)的朋友聊天發(fā)現(xiàn)一個(gè)挺有意思的現(xiàn)象盡管現(xiàn)在實(shí)時(shí)計(jì)算領(lǐng)域新框架層出不窮比如Flink風(fēng)頭正勁但在很多公司的生產(chǎn)環(huán)境里Spark Streaming依然穩(wěn)穩(wěn)地占據(jù)著一席之地處理著大量的實(shí)時(shí)數(shù)據(jù)流。這讓我想起了自己幾年前第一次接觸Spark Streaming的場(chǎng)景當(dāng)時(shí)為了搞定一個(gè)簡(jiǎn)單的實(shí)時(shí)點(diǎn)擊流統(tǒng)計(jì)折騰了好幾個(gè)晚上。現(xiàn)在回過(guò)頭看Spark Streaming的設(shè)計(jì)理念其實(shí)非常經(jīng)典它把流處理巧妙地“偽裝”成了一系列連續(xù)的微批次Micro-Batch處理這種“流批一體”的早期思想讓很多熟悉Spark批處理Spark Core的開(kāi)發(fā)者能夠幾乎無(wú)門(mén)檻地上手實(shí)時(shí)計(jì)算。簡(jiǎn)單來(lái)說(shuō)Spark Streaming是Apache Spark生態(tài)系統(tǒng)里用于處理實(shí)時(shí)數(shù)據(jù)流的組件。它的核心能力是能夠從Kafka、Flume、Kinesis或者TCP Socket等多種數(shù)據(jù)源接入高速數(shù)據(jù)流然后利用Spark強(qiáng)大的分布式計(jì)算引擎對(duì)這些數(shù)據(jù)進(jìn)行高吞吐、可容錯(cuò)的實(shí)時(shí)處理最后將結(jié)果輸出到文件系統(tǒng)、數(shù)據(jù)庫(kù)或者實(shí)時(shí)儀表盤(pán)。它解決的痛點(diǎn)很明確在數(shù)據(jù)產(chǎn)生的瞬間就進(jìn)行分析和響應(yīng)而不是等到攢夠一批再處理這對(duì)于監(jiān)控、風(fēng)控、實(shí)時(shí)推薦等場(chǎng)景至關(guān)重要。那么誰(shuí)適合深入了解一下Spark Streaming呢如果你已經(jīng)是Spark批處理的用戶(hù)想將業(yè)務(wù)擴(kuò)展到實(shí)時(shí)領(lǐng)域那么Spark Streaming是你的自然選擇學(xué)習(xí)曲線非常平緩。如果你在評(píng)估實(shí)時(shí)計(jì)算框架需要的是一個(gè)成熟、穩(wěn)定、社區(qū)資源豐富并且能與現(xiàn)有Spark批處理作業(yè)無(wú)縫整合的方案Spark Streaming也值得你重點(diǎn)考察。當(dāng)然對(duì)于初學(xué)者而言理解Spark Streaming的微批次模型也是理解現(xiàn)代流處理編程范式的絕佳起點(diǎn)。接下來(lái)我會(huì)結(jié)合自己踩過(guò)的坑和積累的經(jīng)驗(yàn)帶你從設(shè)計(jì)思路到實(shí)操細(xì)節(jié)徹底搞懂Spark Streaming。2. 核心架構(gòu)與微批次模型深度解析2.1 DStream流計(jì)算的核心抽象Spark Streaming的編程模型核心是離散化流也就是DStream。這是理解其一切行為的關(guān)鍵。很多新手會(huì)困惑為什么我的流處理作業(yè)延遲感覺(jué)不像Flink那么“實(shí)時(shí)”答案就藏在DStream的設(shè)計(jì)里。你可以把DStream想象成一個(gè)連續(xù)不斷的“數(shù)據(jù)序列”但這個(gè)序列不是平滑的而是被切成了一個(gè)個(gè)固定時(shí)間間隔的“數(shù)據(jù)切片”。每一個(gè)切片本質(zhì)上就是一個(gè)RDD彈性分布式數(shù)據(jù)集。也就是說(shuō)一個(gè)DStream在背后是由一系列按時(shí)間順序排列的RDD所構(gòu)成的。Spark Streaming的作業(yè)調(diào)度器會(huì)周期性地這個(gè)周期就是你設(shè)置的批次間隔比如1秒啟動(dòng)Spark作業(yè)來(lái)處理當(dāng)前時(shí)間窗口內(nèi)到達(dá)的、屬于同一個(gè)RDD的數(shù)據(jù)。舉個(gè)例子你設(shè)置批次間隔為2秒。那么Spark Streaming會(huì)每2秒創(chuàng)建一個(gè)新的RDD這個(gè)RDD包含了這2秒內(nèi)從數(shù)據(jù)源接收到的所有數(shù)據(jù)。然后你定義的所有轉(zhuǎn)換操作如map、filter、reduceByKey都會(huì)作用在這個(gè)RDD上生成新的DStream。這種設(shè)計(jì)帶來(lái)了幾個(gè)深遠(yuǎn)的影響與Spark Core的無(wú)縫繼承所有你在批處理中熟悉的RDD操作、持久化、容錯(cuò)機(jī)制在DStream上幾乎完全適用。你的知識(shí)復(fù)用率極高。一致的語(yǔ)義因?yàn)榈讓邮荝DD所以Spark Streaming能提供“精確一次”的語(yǔ)義保障這對(duì)于金融、交易類(lèi)場(chǎng)景是硬性要求。這通常需要與可靠的數(shù)據(jù)源如Kafka Direct API和可靠的輸出協(xié)同工作。吞吐量?jī)?yōu)先微批次模型天生有利于吞吐量。它可以將一小段時(shí)間內(nèi)的數(shù)據(jù)攢起來(lái)進(jìn)行優(yōu)化后再計(jì)算非常適合高吞吐的日志處理、指標(biāo)聚合場(chǎng)景。注意這個(gè)“批次間隔”是你調(diào)優(yōu)的第一個(gè)關(guān)鍵參數(shù)。設(shè)置得太短如100ms會(huì)導(dǎo)致調(diào)度開(kāi)銷(xiāo)過(guò)大可能每個(gè)批次的數(shù)據(jù)量很小無(wú)法充分發(fā)揮集群性能設(shè)置得太長(zhǎng)如10秒又會(huì)導(dǎo)致數(shù)據(jù)處理延遲變高實(shí)時(shí)性變差。通常在生產(chǎn)環(huán)境中1-5秒是一個(gè)常見(jiàn)的起始探索區(qū)間。2.2 容錯(cuò)與狀態(tài)管理機(jī)制流處理系統(tǒng)必須可靠。Spark Streaming的容錯(cuò)建立在RDD的血統(tǒng)Lineage機(jī)制之上。每個(gè)RDD都知道它是如何從父RDD計(jì)算而來(lái)的。如果某個(gè)節(jié)點(diǎn)宕機(jī)導(dǎo)致某個(gè)RDD分區(qū)丟失Spark可以直接根據(jù)血統(tǒng)重新計(jì)算該分區(qū)從而實(shí)現(xiàn)數(shù)據(jù)恢復(fù)。但對(duì)于有狀態(tài)的計(jì)算例如計(jì)算最近10分鐘的用戶(hù)點(diǎn)擊次數(shù)僅僅重新計(jì)算丟失的數(shù)據(jù)是不夠的因?yàn)闋顟B(tài)本身可能已經(jīng)累積了很久。為此Spark Streaming引入了檢查點(diǎn)機(jī)制和狀態(tài)DStream。檢查點(diǎn)有兩種類(lèi)型。元數(shù)據(jù)檢查點(diǎn)將流計(jì)算應(yīng)用的DAG信息、配置等持久化到HDFS等可靠存儲(chǔ)用于驅(qū)動(dòng)程序的故障恢復(fù)。如果你的Driver程序掛掉重啟后可以從檢查點(diǎn)恢復(fù)上下文并繼續(xù)處理。數(shù)據(jù)檢查點(diǎn)將中間生成的RDD定期保存。這對(duì)于那些血統(tǒng)鏈過(guò)長(zhǎng)例如使用了updateStateByKey且窗口很大的DStream尤為重要可以切斷過(guò)長(zhǎng)的依賴(lài)鏈避免恢復(fù)時(shí)重新計(jì)算整個(gè)歷史。狀態(tài)管理對(duì)于需要跨批次維護(hù)狀態(tài)的操作早期主要使用updateStateByKey。它允許你為每個(gè)Key維護(hù)一個(gè)任意類(lèi)型的狀態(tài)并在每個(gè)批次更新它。但這個(gè)方法有個(gè)問(wèn)題它會(huì)在每個(gè)批次都對(duì)所有Key進(jìn)行計(jì)算即使這個(gè)Key在本批次沒(méi)有新數(shù)據(jù)這在小批次間隔下會(huì)帶來(lái)不小的開(kāi)銷(xiāo)。后來(lái)Spark引入了更高效的mapWithStateAPI。它只對(duì)那些在本批次有更新的Key進(jìn)行狀態(tài)更新和輸出性能提升非常顯著。在最新的Structured Streaming中狀態(tài)管理得到了進(jìn)一步的抽象和優(yōu)化。2.3 與Structured Streaming的關(guān)系辨析這是當(dāng)前Spark流處理生態(tài)中一個(gè)必須厘清的概念。Structured Streaming是Spark 2.0后引入的新的流處理引擎它不再基于DStream而是基于Spark SQL引擎將數(shù)據(jù)流視為一張無(wú)限增長(zhǎng)的表。特性Spark Streaming (DStreams)Structured Streaming編程模型基于RDD的底層API基于DataFrame/Dataset的高級(jí)APIAPI級(jí)別相對(duì)底層靈活性高聲明式更高級(jí)更簡(jiǎn)潔時(shí)間語(yǔ)義主要處理處理時(shí)間原生支持事件時(shí)間、處理時(shí)間以及延遲數(shù)據(jù)的處理水位線不支持支持用于處理亂序事件狀態(tài)管理updateStateByKey/mapWithState內(nèi)建支持更簡(jiǎn)單容錯(cuò)語(yǔ)義可達(dá)到精確一次端到端精確一次需配合特定Source/Sink與批處理統(tǒng)一共享RDD API共享DataFrame API真正做到代碼統(tǒng)一如何選擇對(duì)于新項(xiàng)目強(qiáng)烈建議優(yōu)先考慮Structured Streaming。它在易用性、時(shí)間語(yǔ)義支持和與批處理的統(tǒng)一性上優(yōu)勢(shì)明顯。那為什么還要學(xué)DStream呢首先大量遺留系統(tǒng)仍在運(yùn)行Spark Streaming維護(hù)和優(yōu)化需要相關(guān)知識(shí)。其次DStream API讓你更接近底層對(duì)于理解流計(jì)算的本質(zhì)、進(jìn)行一些極其定制化的操作雖然很少需要仍有價(jià)值。最后學(xué)習(xí)DStream的微批次模型能幫你更好地理解Structured Streaming在底層是如何工作的。3. 從零到一一個(gè)完整的Spark Streaming應(yīng)用實(shí)戰(zhàn)理論說(shuō)得再多不如動(dòng)手跑一遍。我們來(lái)實(shí)現(xiàn)一個(gè)經(jīng)典的場(chǎng)景從Kafka讀取用戶(hù)行為日志JSON格式實(shí)時(shí)統(tǒng)計(jì)每10秒內(nèi)每個(gè)頁(yè)面的訪問(wèn)量PV并將結(jié)果輸出到控制臺(tái)和MySQL數(shù)據(jù)庫(kù)。3.1 環(huán)境準(zhǔn)備與依賴(lài)配置首先你需要一個(gè)Spark環(huán)境。本地測(cè)試最簡(jiǎn)單的方式是下載Spark預(yù)編譯包解壓即可。生產(chǎn)環(huán)境則通常部署在YARN或Kubernetes上。我們假設(shè)使用本地模式進(jìn)行演示。創(chuàng)建一個(gè)標(biāo)準(zhǔn)的Maven或SBT項(xiàng)目。關(guān)鍵的依賴(lài)包括!-- Spark Streaming 核心 -- dependency groupIdorg.apache.spark/groupId artifactIdspark-streaming_2.12/artifactId version3.3.0/version !-- 請(qǐng)使用與Spark Core一致的版本 -- /dependency !-- 用于連接Kafka -- dependency groupIdorg.apache.spark/groupId artifactIdspark-streaming-kafka-0-10_2.12/artifactId version3.3.0/version /dependency !-- MySQL連接器用于輸出 -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.33/version /dependency實(shí)操心得依賴(lài)的Scala版本這里是2.12必須與你安裝的Spark運(yùn)行時(shí)版本嚴(yán)格一致否則會(huì)引發(fā)各種詭異的NoSuchMethodError。最好通過(guò)spark-shell --version命令確認(rèn)你的Spark環(huán)境版本。3.2 應(yīng)用主邏輯編寫(xiě)下面是完整的Scala應(yīng)用示例。我們使用Kafka的Direct API無(wú)Receiver模式這是目前推薦的方式具有更好的并行度和一致性語(yǔ)義。import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.kafka010._ import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe import org.json4s._ import org.json4s.jackson.JsonMethods._ import java.sql.{Connection, DriverManager, PreparedStatement} import java.util.Properties object RealtimePageViewCounter { // 隱式參數(shù)用于json4s解析 implicit val formats: DefaultFormats DefaultFormats case class UserLog(userId: String, pageId: String, timestamp: Long) def main(args: Array[String]): Unit { // 1. 創(chuàng)建SparkConf和StreamingContext批次間隔設(shè)為2秒 val sparkConf new SparkConf() .setAppName(RealtimePageViewCounter) .setMaster(local[2]) // 本地測(cè)試用2個(gè)核生產(chǎn)環(huán)境去掉此參數(shù)通過(guò)spark-submit指定 .set(spark.serializer, org.apache.spark.serializer.KryoSerializer) // 使用Kryo序列化提升性能 val ssc new StreamingContext(sparkConf, Seconds(2)) // 設(shè)置檢查點(diǎn)目錄用于狀態(tài)恢復(fù)本地測(cè)試可先注釋 // ssc.checkpoint(hdfs://your-nn:9000/spark-streaming-checkpoint) // 2. 配置Kafka參數(shù) val kafkaParams Map[String, Object]( bootstrap.servers - kafka-broker1:9092,kafka-broker2:9092, key.deserializer - org.apache.kafka.common.serialization.StringDeserializer, value.deserializer - org.apache.kafka.common.serialization.StringDeserializer, group.id - spark-streaming-pageview-group, auto.offset.reset - latest, // 從最新位置開(kāi)始消費(fèi) enable.auto.commit - (false: java.lang.Boolean) // Spark自己管理offset ) val topics Array(user-behavior-topic) // 3. 創(chuàng)建DStream連接Kafka val stream KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams) ) // 4. 數(shù)據(jù)處理邏輯 val pageCounts stream .map(record record.value()) // 提取Kafka消息的值JSON字符串 .filter(_.nonEmpty) // 過(guò)濾空消息 .map { jsonString try { // 解析JSON提取pageId val json parse(jsonString) val pageId (json \ pageId).extractOrElse[String](unknown) (pageId, 1) } catch { case e: Exception // 記錄解析錯(cuò)誤實(shí)際生產(chǎn)中應(yīng)寫(xiě)入錯(cuò)誤日志或死信隊(duì)列 println(sFailed to parse JSON: $jsonString, error: ${e.getMessage}) (parse_error, 1) } } .reduceByKeyAndWindow( _ _, // 聚合函數(shù)累加 _ - _, // 逆函數(shù)用于窗口滑動(dòng)時(shí)減去過(guò)期批次提升性能需設(shè)置檢查點(diǎn) Seconds(10), // 窗口長(zhǎng)度10秒 Seconds(2) // 滑動(dòng)間隔2秒與批次間隔相同 ) // 如果不使用逆函數(shù)可以用簡(jiǎn)單的 reduceByKey(_ _).window(Seconds(10), Seconds(2)) // 5. 輸出操作觸發(fā)計(jì)算并輸出 pageCounts.foreachRDD { (rdd, time) // 注意foreachRDD內(nèi)部的代碼在Driver端執(zhí)行但其中的RDD操作在Executor端執(zhí)行 if (!rdd.isEmpty()) { println(s\n Batch Time: $time ) // 輸出到控制臺(tái) rdd.foreachPartition { partitionOfRecords // 這個(gè)foreach在Executor上執(zhí)行 partitionOfRecords.foreach { case (pageId, count) println(sPage: $pageId, Count: $count) } } // 輸出到MySQL (在Driver端收集少量數(shù)據(jù)后寫(xiě)入或使用foreachPartition在Executor寫(xiě)) // 方式A收集到Driver后寫(xiě)入適合結(jié)果集小 val collectedData rdd.collect() if (collectedData.nonEmpty) { saveToMySQL(collectedData, time) } // 方式B使用foreachPartition在Executor分布式寫(xiě)入適合結(jié)果集大但需管理連接池 // rdd.foreachPartition { partition // val conn getMySQLConnection() // // ... 批量插入邏輯 // conn.close() // } } } // 6. 啟動(dòng)流計(jì)算并等待終止 ssc.start() ssc.awaitTermination() } def saveToMySQL(data: Array[(String, Int)], batchTime: org.apache.spark.streaming.Time): Unit { var conn: Connection null var pstmt: PreparedStatement null val url jdbc:mysql://your-mysql-host:3306/streaming_db val user your_user val password your_password try { Class.forName(com.mysql.cj.jdbc.Driver) conn DriverManager.getConnection(url, user, password) // 假設(shè)表結(jié)構(gòu)page_pv (batch_time TIMESTAMP, page_id VARCHAR(50), pv INT) val sql INSERT INTO page_pv (batch_time, page_id, pv) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE pv ? pstmt conn.prepareStatement(sql) val timestamp new java.sql.Timestamp(batchTime.milliseconds) for ((pageId, count) - data) { pstmt.setTimestamp(1, timestamp) pstmt.setString(2, pageId) pstmt.setInt(3, count) pstmt.setInt(4, count) // 用于ON DUPLICATE KEY UPDATE pstmt.addBatch() } pstmt.executeBatch() println(sSuccessfully saved ${data.length} records to MySQL.) } catch { case e: Exception e.printStackTrace() } finally { if (pstmt ! null) pstmt.close() if (conn ! null) conn.close() } } }3.3 關(guān)鍵代碼段解析與調(diào)優(yōu)點(diǎn)StreamingContext初始化這是所有流計(jì)算的起點(diǎn)。Seconds(2)定義了微批次的間隔。local[2]中的數(shù)字2代表至少使用2個(gè)CPU核心一個(gè)用于接收數(shù)據(jù)一個(gè)用于處理數(shù)據(jù)這是本地測(cè)試的最低要求。Kafka Direct API我們使用createDirectStream并設(shè)置enable.auto.commit為false。這意味著Spark Streaming會(huì)自己將消費(fèi)偏移量offset管理在檢查點(diǎn)中或自己提交回Kafka這是實(shí)現(xiàn)“精確一次”處理的基礎(chǔ)。你需要確保輸出操作是冪等的或者將offset和輸出結(jié)果放在同一個(gè)事務(wù)中。reduceByKeyAndWindow這是窗口操作的核心。我們?cè)O(shè)置了10秒的窗口長(zhǎng)度和2秒的滑動(dòng)間隔。這意味著每2秒一個(gè)批次我們會(huì)計(jì)算過(guò)去10秒內(nèi)的數(shù)據(jù)。使用了加法和減法函數(shù)這要求開(kāi)啟檢查點(diǎn)但能極大優(yōu)化滑動(dòng)窗口的性能因?yàn)樗恍枰貜?fù)計(jì)算重疊部分的數(shù)據(jù)。foreachRDD的設(shè)計(jì)模式這是輸出結(jié)果到外部系統(tǒng)如數(shù)據(jù)庫(kù)、Redis的標(biāo)準(zhǔn)入口。至關(guān)重要的一點(diǎn)foreachRDD內(nèi)部的代碼在Driver端執(zhí)行但其中的RDD操作如foreachPartition是在Executor端執(zhí)行的。創(chuàng)建數(shù)據(jù)庫(kù)連接等昂貴操作應(yīng)該在foreachPartition內(nèi)部進(jìn)行并為每個(gè)分區(qū)創(chuàng)建一個(gè)連接池而不是為每條記錄創(chuàng)建連接更不要在Driver端創(chuàng)建連接然后序列化到Executor這會(huì)導(dǎo)致序列化錯(cuò)誤。4. 生產(chǎn)環(huán)境部署與性能調(diào)優(yōu)指南把應(yīng)用跑起來(lái)只是第一步要讓它在生產(chǎn)環(huán)境中穩(wěn)定、高效地運(yùn)行還需要做大量工作。4.1 資源分配與并行度優(yōu)化Spark Streaming應(yīng)用的性能很大程度上取決于資源是否給夠以及任務(wù)是否被充分并行化。Executor資源通過(guò)spark-submit提交時(shí)需要合理設(shè)置。--num-executorsExecutor數(shù)量。根據(jù)數(shù)據(jù)量和處理邏輯復(fù)雜度決定通常從10-20個(gè)開(kāi)始。--executor-cores每個(gè)Executor的CPU核心數(shù)。建議2-4個(gè)確保每個(gè)Executor能并行執(zhí)行多個(gè)任務(wù)。--executor-memory每個(gè)Executor的內(nèi)存。需要容納接收到的批次數(shù)據(jù)、進(jìn)行轉(zhuǎn)換操作產(chǎn)生的中間數(shù)據(jù)以及維護(hù)的狀態(tài)。必須預(yù)留一部分給操作系統(tǒng)和HDFS客戶(hù)端約10%。例如總內(nèi)存4G可設(shè)置--executor-memory 3g。Receiver與并行度如果使用舊的Receiver模式不推薦接收數(shù)據(jù)本身會(huì)占用一個(gè)CPU核心。在Direct API下Kafka分區(qū)數(shù)直接決定了讀取階段的并行度。確保Kafka主題的分區(qū)數(shù) Spark Streaming作業(yè)中讀取該主題的并發(fā)任務(wù)數(shù)。通常你可以通過(guò)spark.streaming.kafka.maxRatePerPartition參數(shù)控制每個(gè)分區(qū)每秒讀取的最大消息數(shù)來(lái)平衡吞吐和延遲。處理并行度由RDD的分區(qū)數(shù)決定。Shuffle操作如reduceByKey后的默認(rèn)分區(qū)數(shù)由spark.default.parallelism控制通常設(shè)置為executor-cores * num-executors的2-3倍。你也可以在操作中顯式指定分區(qū)數(shù)如reduceByKey(__, 100)。4.2 背壓機(jī)制與動(dòng)態(tài)資源分配當(dāng)數(shù)據(jù)流入速度超過(guò)處理速度時(shí)會(huì)導(dǎo)致批次處理時(shí)間越來(lái)越長(zhǎng)最終堆積崩潰。Spark Streaming 1.5之后引入了背壓機(jī)制可以動(dòng)態(tài)調(diào)整接收速率來(lái)適配處理能力。啟用背壓設(shè)置spark.streaming.backpressure.enabledtrue。背壓算法默認(rèn)使用PID控制器你也可以通過(guò)spark.streaming.backpressure.initialRate設(shè)置初始接收速率。啟用后Spark會(huì)監(jiān)控批次處理時(shí)間和調(diào)度延遲自動(dòng)調(diào)整從Kafka等源拉取數(shù)據(jù)的速率。對(duì)于運(yùn)行在YARN上的應(yīng)用還可以結(jié)合動(dòng)態(tài)資源分配。但這在流處理中需謹(jǐn)慎使用因?yàn)樯暾?qǐng)和釋放Executor需要時(shí)間可能影響實(shí)時(shí)性。通常適用于處理負(fù)載有明顯波峰波谷且對(duì)延遲不極度敏感的場(chǎng)景。4.3 檢查點(diǎn)與狀態(tài)恢復(fù)實(shí)戰(zhàn)檢查點(diǎn)不是可選項(xiàng)對(duì)于生產(chǎn)應(yīng)用是必選項(xiàng)。它用于元數(shù)據(jù)恢復(fù)和狀態(tài)計(jì)算。設(shè)置檢查點(diǎn)目錄目錄必須是一個(gè)可靠的文件系統(tǒng)如HDFS。ssc.checkpoint(“hdfs://...”。編寫(xiě)可恢復(fù)的驅(qū)動(dòng)程序你的主函數(shù)需要能被Spark在故障后重新調(diào)用。標(biāo)準(zhǔn)模式如下def createStreamingContext(): StreamingContext { val sparkConf ... val ssc new StreamingContext(sparkConf, Seconds(2)) // 定義你的DStream計(jì)算邏輯 val lines ... // ... ssc.checkpoint(checkpointDir) ssc } def main(args: Array[String]) { val checkpointDir “hdfs://...” val ssc StreamingContext.getOrCreate(checkpointDir, createStreamingContext _) ssc.start() ssc.awaitTermination() }這樣當(dāng)Driver重啟時(shí)getOrCreate會(huì)嘗試從檢查點(diǎn)目錄重建StreamingContext。如果失敗則調(diào)用提供的函數(shù)創(chuàng)建新的。踩坑實(shí)錄檢查點(diǎn)目錄包含了序列化的類(lèi)。如果你修改了應(yīng)用代碼如添加了新的類(lèi)字段然后試圖從舊的檢查點(diǎn)恢復(fù)會(huì)引發(fā)序列化錯(cuò)誤。最佳實(shí)踐是每次代碼升級(jí)后清空檢查點(diǎn)目錄意味著從最新的Kafka偏移量開(kāi)始消費(fèi)或者確保代碼變更向后兼容。5. 典型問(wèn)題排查與監(jiān)控運(yùn)維即使應(yīng)用部署成功運(yùn)維過(guò)程中也會(huì)遇到各種問(wèn)題。這里記錄幾個(gè)最常見(jiàn)的問(wèn)題和排查思路。5.1 批次處理延遲與堆積這是最常見(jiàn)的問(wèn)題。癥狀是Spark UI的Streaming頁(yè)面上批次處理時(shí)間Processing Time持續(xù)大于批次間隔Batch Interval導(dǎo)致“Scheduling Delay”不斷增長(zhǎng)。排查步驟看日志首先查看Executor和Driver的日志是否有明顯的錯(cuò)誤或GC警告。看Spark UIStreaming頁(yè)確認(rèn)哪些批次延遲了。是持續(xù)延遲還是偶發(fā)Stages頁(yè)點(diǎn)擊延遲批次對(duì)應(yīng)的作業(yè)查看是哪個(gè)Stage耗時(shí)最長(zhǎng)。是讀取數(shù)據(jù)慢Shuffle慢還是輸出慢Executors頁(yè)觀察GC時(shí)間是否過(guò)長(zhǎng)。如果Full GC頻繁說(shuō)明內(nèi)存不足。針對(duì)性?xún)?yōu)化數(shù)據(jù)傾斜如果某個(gè)Stage的某個(gè)Task執(zhí)行時(shí)間遠(yuǎn)長(zhǎng)于其他很可能是數(shù)據(jù)傾斜。使用sample方法查看Key分布考慮使用加鹽隨機(jī)前綴打散熱點(diǎn)Key。外部系統(tǒng)瓶頸如果延遲發(fā)生在foreachRDD的輸出階段可能是數(shù)據(jù)庫(kù)或Redis寫(xiě)入慢?？紤]使用連接池、批量寫(xiě)入、異步寫(xiě)入或換用更高性能的輸出端。資源不足如果所有Task都慢且GC正?？赡苁荂PU或內(nèi)存整體不足。嘗試增加executor-cores或executor-memory。調(diào)整批次間隔適當(dāng)增大批次間隔如從1秒到2秒給每個(gè)批次更多處理時(shí)間可以緩解短期壓力但會(huì)犧牲實(shí)時(shí)性。5.2 數(shù)據(jù)丟失與重復(fù)消費(fèi)這通常與偏移量管理和輸出操作的原子性有關(guān)。確保精確一次語(yǔ)義使用Direct API它讓Spark自己管理Kafka偏移量?？煽康臄?shù)據(jù)源確保Kafka本身是高可用的。冪等的輸出或事務(wù)性輸出這是最難的部分。要么你的輸出操作是冪等的比如INSERT ON DUPLICATE KEY UPDATE要么你將偏移量的提交和數(shù)據(jù)的輸出放在同一個(gè)數(shù)據(jù)庫(kù)事務(wù)中。Spark本身不提供跨系統(tǒng)的事務(wù)這需要你在foreachRDD中自己實(shí)現(xiàn)。監(jiān)控偏移量定期檢查Spark提交到Kafka的消費(fèi)者組偏移量確保其正常推進(jìn)并與實(shí)際處理進(jìn)度匹配。5.3 監(jiān)控與告警體系搭建不能等用戶(hù)投訴了才發(fā)現(xiàn)流處理作業(yè)掛了。必須建立監(jiān)控。Spark UI History Server這是最基本的。通過(guò)History Server可以查看已結(jié)束應(yīng)用的運(yùn)行情況。Metrics系統(tǒng)Spark提供了豐富的Metrics可以通過(guò)SparkConf配置輸出到Ganglia、Graphite、Prometheus等系統(tǒng)。關(guān)鍵指標(biāo)包括spark.streaming.*: 如processingDelay處理延遲、schedulingDelay調(diào)度延遲、numReceivers接收器數(shù)量、numTotalCompletedBatches總完成批次數(shù)等。JVM相關(guān)指標(biāo)GC時(shí)間、堆內(nèi)存使用情況。自定義應(yīng)用指標(biāo)你可以在foreachRDD里將每批次處理的數(shù)據(jù)量、輸出記錄數(shù)等業(yè)務(wù)指標(biāo)推送到你的監(jiān)控系統(tǒng)如StatsD。進(jìn)程存活監(jiān)控使用系統(tǒng)級(jí)的監(jiān)控工具如Supervisord、K8s Liveness Probe確保Driver和Executor進(jìn)程存活。對(duì)于YARN可以監(jiān)控YARN Application狀態(tài)。告警規(guī)則針對(duì)關(guān)鍵指標(biāo)設(shè)置告警例如連續(xù)N個(gè)批次處理延遲超過(guò)閾值、消費(fèi)者組滯后Lag持續(xù)增長(zhǎng)、Executor頻繁丟失等。Spark Streaming是一個(gè)經(jīng)歷過(guò)大規(guī)模生產(chǎn)環(huán)境考驗(yàn)的框架它的微批次模型在吞吐量和一致性之間取得了很好的平衡。雖然Structured Streaming代表了未來(lái)的方向但理解DStream的運(yùn)作機(jī)制、掌握其調(diào)優(yōu)和運(yùn)維技巧對(duì)于任何一個(gè)大數(shù)據(jù)開(kāi)發(fā)者來(lái)說(shuō)仍然是一筆寶貴的財(cái)富。在實(shí)際項(xiàng)目中最關(guān)鍵的是根據(jù)業(yè)務(wù)對(duì)延遲和吞吐量的具體要求以及對(duì)一致性的容忍度來(lái)做出最合適的架構(gòu)選擇和技術(shù)決策。