1. 從批處理到流處理為什么Spark Streaming是實時計算的“定海神針”如果你用過Spark做批處理那你一定體驗過它處理海量離線數(shù)據(jù)時那種“力大磚飛”的快感。但數(shù)據(jù)世界不是靜止的業(yè)務對時效性的要求越來越高報表從T1變成小時級再到分鐘級甚至秒級。這時候傳統(tǒng)的批處理框架就顯得有些力不從心了。你可能會想能不能把源源不斷產生的數(shù)據(jù)切成一個個小批次然后用Spark批處理引擎來處理這些小批次呢這個樸素的想法正是Spark Streaming的核心設計哲學。Spark Streaming并不是一個獨立的流處理引擎它更像是Spark核心引擎的一個“流式皮膚”。它把連續(xù)的數(shù)據(jù)流按照你設定的時間間隔比如1秒、5秒切割成一系列微小的、確定性的批處理作業(yè)也就是所謂的“微批次”。然后它利用Spark強大的分布式計算能力并行處理這些微批次。這種架構帶來的最大好處就是批流統(tǒng)一。你寫的Spark代碼無論是RDD的轉換操作還是DataFrame的SQL查詢在Spark Streaming里幾乎可以無縫遷移。這意味著團隊的學習成本和代碼維護成本大大降低你不需要為了實時計算再去學習一套全新的、復雜的API。我見過不少團隊在選型實時計算框架時在Spark Streaming和Flink之間反復糾結。Flink是真正的逐事件處理引擎延遲可以做到毫秒級理論上的確更“實時”。但Spark Streaming的微批次模型在秒到分鐘的延遲級別上其穩(wěn)定性、Exactly-Once語義的成熟度以及與Spark生態(tài)如MLlib機器學習庫、GraphX圖計算的無縫集成讓它成為了許多中高延遲實時場景的“定海神針”。比如實時儀表盤、網(wǎng)絡監(jiān)控、實時ETL、以及一些對延遲要求不是極端苛刻的實時推薦場景Spark Streaming憑借其簡單、穩(wěn)健和生態(tài)完整的特性往往是更務實的選擇。2. DStream理解Spark Streaming編程模型的第一塊基石要玩轉Spark Streaming你必須先吃透它的核心抽象——DStream。你可以把DStream理解為一個連續(xù)不斷的RDD序列。每個RDD包含了一個特定時間間隔內到達的所有數(shù)據(jù)。假設你設置批次間隔為2秒那么每過2秒這段時間內流入的數(shù)據(jù)就會被包裝成一個RDD然后這個RDD被追加到DStream這個序列的末尾。這種設計非常巧妙它讓流處理變得像批處理一樣直觀。所有你在Spark Core里學到的RDD操作比如map、filter、reduceByKey在DStream上都有對應的版本。當你對一個DStream調用map時這個操作會作用在DStream底層的每一個RDD上。例如你有一個包含一行行日志的DStream通過map操作可以輕松地將每一行日志拆分成單詞。// 假設 lines 是一個DStream[String]每行是一條日志 val words lines.flatMap(_.split( ))這里的關鍵在于words本身也是一個DStream。轉換操作并不會立即執(zhí)行它們只是被記錄了下來構建了一個有向無環(huán)圖。只有當輸出操作如print、saveAsTextFiles被調用時整個計算邏輯才會被觸發(fā)并由Spark調度器分配到集群上執(zhí)行。這種惰性求值的機制和Spark批處理是一脈相承的。DStream API分為兩大類轉換和輸出。轉換操作產生新的DStream而輸出操作則將數(shù)據(jù)推送到外部系統(tǒng)如數(shù)據(jù)庫、文件系統(tǒng)或打印到控制臺這才是真正觸發(fā)計算的“動作”。此外DStream還提供了一些針對流處理場景的特殊轉換比如window和reduceByKeyAndWindow用于滑動窗口計算這是實時聚合統(tǒng)計的利器。理解DStream是離散化的流這個本質是寫出正確、高效Spark Streaming程序的基礎。3. 實戰(zhàn)入門構建你的第一個Spark Streaming應用理論說再多不如動手跑一遍。我們來構建一個最簡單的網(wǎng)絡詞頻統(tǒng)計應用它監(jiān)聽一個網(wǎng)絡端口對接收到的文本行進行實時單詞計數(shù)。這個例子雖小但涵蓋了初始化、輸入、轉換、輸出整個閉環(huán)。首先你需要準備環(huán)境。確保你有一個可以運行的Spark環(huán)境無論是本地模式還是集群模式。對于本地測試最簡單的就是使用本地模式并設置至少2個線程一個用于接收數(shù)據(jù)一個用于處理計算。import org.apache.spark._ import org.apache.spark.streaming._ // 1. 創(chuàng)建StreamingContext這是所有Spark Streaming功能的入口點 // 參數(shù)SparkConf對象 批次間隔時間例如Seconds(2) val conf new SparkConf().setAppName(NetworkWordCount).setMaster(local[2]) val ssc new StreamingContext(conf, Seconds(2)) // 2. 創(chuàng)建輸入DStream。這里從TCP Socket源讀取數(shù)據(jù) // 參數(shù)主機名 端口號 val lines ssc.socketTextStream(localhost, 9999) // 3. 轉換操作將行拆分為單詞然后計數(shù) val words lines.flatMap(_.split( )) val pairs words.map(word (word, 1)) val wordCounts pairs.reduceByKey(_ _) // 4. 輸出操作打印每個批次的前10個記錄到控制臺 wordCounts.print() // 5. 啟動流式計算 ssc.start() // 6. 等待計算被終止手動或錯誤 ssc.awaitTermination()代碼邏輯很清晰。但這里有幾個新手極易踩坑的細節(jié)關于local[2]在本地模式下local[2]表示使用2個線程。為什么至少是2因為Spark Streaming需要一個線程來接收數(shù)據(jù)Receiver另一個線程來處理數(shù)據(jù)Job。如果只設置local[1]Receiver會獨占這個線程導致處理作業(yè)永遠無法執(zhí)行程序看似啟動了但不會有任何輸出。這是第一個“坑”。關于socketTextStream這是一個用于測試和演示的源。在生產環(huán)境中你幾乎不會用它。但它非常適合入門因為你可以用簡單的nc -lk 9999命令啟動一個網(wǎng)絡服務來發(fā)送數(shù)據(jù)直觀地看到流處理的效果。關于ssc.start()和awaitTermination()start()是啟動引擎開始調度作業(yè)。awaitTermination()則讓主線程阻塞在這里等待計算結束。沒有awaitTermination()主線程會立刻結束導致整個應用退出。這是第二個容易忘記的“坑”。運行這個程序前你需要在終端用nc -lk 9999命令啟動一個Netcat服務器。然后在另一個終端啟動你的Spark應用。之后在Netcat終端輸入一些句子你就能在Spark應用的控制臺看到實時統(tǒng)計出的詞頻了。這個過程能讓你最直觀地感受到“流”是如何被切成“批”并被處理的。4. 核心概念深潛批次間隔、窗口與狀態(tài)管理當你跑通第一個例子后有三個核心概念必須深入理解它們決定了你程序的性能、延遲和功能邊界。4.1 批次間隔吞吐量與延遲的權衡批次間隔是你創(chuàng)建StreamingContext時設定的時間參數(shù)如Seconds(5)。它有兩個核心影響數(shù)據(jù)處理延遲數(shù)據(jù)最快也要在一個批次間隔結束后才能被處理。因此理論上最小延遲等于批次間隔。設為1秒延遲就在1秒左右設為5秒延遲就在5秒左右。系統(tǒng)吞吐量更小的批次間隔意味著更頻繁地調度作業(yè)這會增加調度開銷。如果每個批次的數(shù)據(jù)量很小但調度很頻繁集群資源可能浪費在啟動作業(yè)上反而降低了吞吐量。如何選擇沒有一個黃金值。你需要根據(jù)業(yè)務對延遲的容忍度和數(shù)據(jù)到達速率來權衡。一個實用的方法是先從較大的間隔如10-30秒開始觀察集群資源使用率和處理延遲。如果資源利用率不高且延遲允許再逐步調小間隔直到找到吞吐量和延遲的平衡點。監(jiān)控Spark UI中的“Streaming”標簽頁關注“Processing Delay”處理延遲和“Scheduling Delay”調度延遲如果調度延遲持續(xù)增長說明批次處理時間已經(jīng)超過了批次間隔系統(tǒng)正在積壓這時你需要考慮調大間隔或優(yōu)化程序/增加資源。4.2 窗口操作處理時間滑動的數(shù)據(jù)塊很多實時統(tǒng)計需求不是針對單個批次的而是針對最近一段時間的數(shù)據(jù)比如“過去5分鐘內熱門搜索詞”、“最近1小時網(wǎng)站獨立訪客數(shù)”。這就是窗口操作的用武之地。窗口操作需要兩個參數(shù)窗口長度窗口覆蓋的時間長度比如Minutes(5)?；瑒娱g隔窗口每次向前滑動的時間間隔比如Minutes(1)。// 每2秒一個批次計算過去10秒內的單詞計數(shù)每4秒更新一次結果 val windowedWordCounts pairs.reduceByKeyAndWindow( (a: Int, b: Int) a b, // 添加新進入窗口的批次 (a: Int, b: Int) a - b, // 移除舊滑出窗口的批次可選用于優(yōu)化 Seconds(10), // 窗口長度 Seconds(4) // 滑動間隔 )這里有一個關鍵點窗口長度和滑動間隔必須是批次間隔的整數(shù)倍。滑動間隔決定了結果DStream的輸出頻率。上面的例子每4秒會輸出一個過去10秒的統(tǒng)計結果。使用窗口操作時內存和計算開銷會增大因為系統(tǒng)需要保存多個批次的數(shù)據(jù)。提供“逆函數(shù)”上面代碼中的減法是可選的但Spark可以利用它進行增量計算只計算新進入和滑出窗口的數(shù)據(jù)差異而不是在每次滑動時都對窗口內所有數(shù)據(jù)全量重算這能極大提升效率。4.3 狀態(tài)管理實現(xiàn)跨批次的記憶有些場景需要記住之前批次的信息比如統(tǒng)計從流開始以來的總單詞數(shù)或者跟蹤一個用戶會話的狀態(tài)。這就需要用到有狀態(tài)轉換主要API是updateStateByKey或更高效的mapWithState。updateStateByKey允許你為每個鍵如單詞維護一個任意類型的狀態(tài)如累計計數(shù)并在每個新批次到來時用一個函數(shù)更新這個狀態(tài)。// 定義一個更新函數(shù)將新值加到舊狀態(tài)上 def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] { Some(runningCount.getOrElse(0) newValues.sum) } // 使用updateStateByKey進行全局詞頻統(tǒng)計 val runningCounts pairs.updateStateByKey[Int](updateFunction _)狀態(tài)管理是把雙刃劍。它功能強大但意味著Spark需要將每個鍵的狀態(tài)保存在內存或溢寫到磁盤中并定期向HDFS等可靠存儲做檢查點備份以保證容錯性。如果你的鍵空間非常大例如數(shù)億個不同的用戶ID狀態(tài)管理會消耗巨大的內存并可能成為性能瓶頸。因此使用前務必評估狀態(tài)的規(guī)模和生命周期對于無限增長的鍵空間可能需要設計TTL生存時間機制來清理舊狀態(tài)。5. 輸入源與輸出操作連接真實世界的橋梁DStream的轉換操作是在Spark內存中進行的數(shù)據(jù)舞蹈而輸入源和輸出操作則是連接外部數(shù)據(jù)世界的大門。選擇不當大門就會成為瓶頸。5.1 輸入源數(shù)據(jù)從何而來Spark Streaming支持多種輸入源可分為兩類基礎源直接存在于StreamingContext API中的源如文件系統(tǒng)、Socket。socketTextStream就屬于此類僅用于測試。高級源需要通過額外工具類連接的源如Kafka、Flume、Kinesis。這些是生產環(huán)境的主流選擇。對于生產環(huán)境Apache Kafka是事實上的標準選擇。Spark Streaming提供了兩種連接Kafka的方式Receiver-based Approach使用一個Receiver線程預拉取數(shù)據(jù)到Spark內存中并做WALWrite Ahead Log保證數(shù)據(jù)不丟。這種方式存在內存壓力大和可能的數(shù)據(jù)重復問題已逐漸被淘汰。Direct Approach (推薦)Spark Streaming每個批次周期直接去Kafka對應分區(qū)拉取指定偏移量范圍的數(shù)據(jù)。這種方式更高效具有更好的端到端Exactly-Once語義且無需WAL簡化了架構。你需要使用KafkaUtils.createDirectStream這個API。import org.apache.spark.streaming.kafka010._ val kafkaParams Map[String, Object]( bootstrap.servers - broker1:9092,broker2:9092, key.deserializer - classOf[StringDeserializer], value.deserializer - classOf[StringDeserializer], group.id - spark-streaming-group, auto.offset.reset - latest, enable.auto.commit - (false: java.lang.Boolean) // 手動管理偏移量 ) val topics Array(input-topic) val stream KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) ) // 獲取Kafka消息的value部分 val lines stream.map(record record.value())使用Direct方式時偏移量管理至關重要。通常做法是將消費偏移量存儲在如ZooKeeper、Kafka自身__consumer_offsets或自定義數(shù)據(jù)庫如HBase中并在輸出操作完成后異步提交以實現(xiàn)“輸出成功后才提交偏移量”的原子性這是保證Exactly-Once處理語義的關鍵一環(huán)。5.2 輸出操作數(shù)據(jù)去向何方輸出操作是觸發(fā)實際計算的行動算子。除了print()用于調試常用輸出包括saveAsTextFiles/ saveAsObjectFiles/ saveAsHadoopFiles: 保存到HDFS等文件系統(tǒng)。foreachRDD:這是最強大、最靈活的輸出算子它允許你對每個批次的RDD執(zhí)行任意操作。foreachRDD的設計初衷是讓你能夠將數(shù)據(jù)寫入到任何不支持Spark Streaming原生集成的存儲系統(tǒng)中比如MySQL、Redis、Elasticsearch等。但這里有一個超級大坑// 錯誤寫法連接對象在Driver端創(chuàng)建無法序列化到Executor端執(zhí)行。 wordCounts.foreachRDD { rdd val connection createNewDatabaseConnection() // 在Driver創(chuàng)建 rdd.foreach { record connection.send(record) // 嘗試在Executor執(zhí)行會報序列化錯誤 } }// 正確寫法使用rdd.foreachPartition在每個分區(qū)內創(chuàng)建連接。 wordCounts.foreachRDD { rdd rdd.foreachPartition { partitionOfRecords val connection createNewDatabaseConnection() // 在Executor端創(chuàng)建 partitionOfRecords.foreach(record connection.send(record)) connection.close() } }核心原則連接對象如數(shù)據(jù)庫連接、網(wǎng)絡連接的創(chuàng)建必須在Executor端進行即在foreachRDD內部的rdd.foreachPartition或rdd.foreach中并且最好在分區(qū)級別創(chuàng)建以復用連接避免每條記錄都創(chuàng)建/銷毀連接帶來的巨大開銷。更進一步可以使用連接池來優(yōu)化性能。6. 容錯、監(jiān)控與性能調優(yōu)實戰(zhàn)指南一個Spark Streaming應用開發(fā)完成后讓它穩(wěn)定、高效地在生產環(huán)境運行是更大的挑戰(zhàn)。6.1 容錯與檢查點Spark Streaming通過檢查點來實現(xiàn)容錯。主要有兩種類型元數(shù)據(jù)檢查點將定義流計算的有向無環(huán)圖信息保存到HDFS等可靠存儲。用于Driver程序故障恢復后重建StreamingContext。數(shù)據(jù)檢查點將有狀態(tài)轉換如updateStateByKey的中間RDD定期保存。因為狀態(tài)計算鏈可能很長故障恢復時從頭計算代價高檢查點可以切斷這個依賴鏈。啟用檢查點非常簡單ssc.checkpoint(hdfs://path/to/checkpoint-directory)對于有狀態(tài)操作這是必須的。對于無狀態(tài)操作如果你希望Driver故障后能從斷點恢復也需要啟用。這里有個經(jīng)驗檢查點目錄必須是一個像HDFS這樣所有節(jié)點都能訪問的、高可用的分布式文件系統(tǒng)路徑絕對不能是本地路徑。否則節(jié)點故障時數(shù)據(jù)會丟失。6.2 監(jiān)控你的流應用監(jiān)控是生產運維的眼睛。除了通用的Spark UI關注Stages、Storage、Environment等標簽頁要特別關注Streaming專屬的“Streaming”標簽頁Processing Delay處理每個批次實際花費的時間。理想情況應小于批次間隔。Scheduling Delay批次在隊列中等待調度的時間。如果持續(xù)增長說明系統(tǒng)處理不過來批次在積壓。Input Rate/Processing Rate數(shù)據(jù)輸入速率和處理速率。處理速率應持續(xù)高于輸入速率。Total DelayProcessing Delay Scheduling Delay即端到端延遲。如果Scheduling Delay不斷增長就是明確的告警信號。你需要立即行動要么增加資源更多Executor、更大內存要么優(yōu)化程序邏輯要么調大批次間隔。6.3 性能調優(yōu)要點批次間隔與并行度如前所述找到合適的批次間隔。同時通過spark.streaming.blockInterval參數(shù)默認200ms可以控制Receiver將數(shù)據(jù)塊封裝成RDD分區(qū)的時間從而影響并行度。更小的blockInterval會產生更多分區(qū)提高并行度但也會增加任務調度開銷。垃圾回收優(yōu)化Spark Streaming應用因為常駐并產生大量小批次RDD對GC壓力很大。建議使用G1垃圾回收器并在Spark配置中增加相關GC調優(yōu)參數(shù)如-XX:UseG1GC并監(jiān)控GC時間。反壓機制在Spark 1.5版本可以開啟反壓spark.streaming.backpressure.enabledtrue。當系統(tǒng)處理速度跟不上數(shù)據(jù)輸入速度時反壓機制能動態(tài)調整接收速率防止內存溢出。這是一個非常重要的安全閥。數(shù)據(jù)序列化使用Kryo序列化spark.serializerorg.apache.spark.serializer.KryoSerializer替代默認的Java序列化能顯著減少數(shù)據(jù)體積和序列化/反序列化時間。優(yōu)雅關閉使用ssc.stop(stopSparkContexttrue, stopGracefullytrue)可以讓流式應用在處理完當前批次的數(shù)據(jù)后再關閉避免數(shù)據(jù)丟失。結合外部信號如檢測HDFS上的標記文件可以實現(xiàn)計劃內的應用重啟和升級。7. 從DStream到Structured Streaming流處理的演進當你熟練使用DStream API后可能會遇到一些痛點API是低層次的RDD操作需要手動處理容錯語義事件時間處理和支持亂序數(shù)據(jù)比較麻煩與批處理的DataFrame/Dataset API割裂。這正是Spark 2.0引入Structured Streaming的動機。它基于Spark SQL引擎將流數(shù)據(jù)視為一張無限增長的表。你使用熟悉的DataFrame/DataSet API進行聲明式查詢Spark負責以增量的方式持續(xù)更新結果表。它原生支持事件時間、窗口操作、水印處理亂序數(shù)據(jù)并且通過檢查點和預寫日志實現(xiàn)了端到端的Exactly-Once語義API更簡潔概念更統(tǒng)一。// Structured Streaming 實現(xiàn)詞頻統(tǒng)計 val spark SparkSession.builder... val lines spark.readStream.format(socket)... val words lines.as[String].flatMap(_.split( )) val wordCounts words.groupBy(value).count() val query wordCounts.writeStream .outputMode(complete) // 或 append, update .format(console) .start()對于新項目強烈建議優(yōu)先考慮Structured Streaming。它代表了Spark流處理的未來方向社區(qū)投入也更大。DStream API目前處于維護模式。不過理解DStream的微批次模型和底層原理對于你深入掌握Structured Streaming乃至排查復雜問題仍然有不可替代的價值。它讓你明白再高級的抽象最終也是構建在核心的批處理引擎和離散化流的思想之上。