Spark流式編程介紹 - 編程模型

• 2019 年 10 月 3 日
• 筆記

來源Spark官方文檔
http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#programming-model

編程模型

結構化流中的核心概念就是將活動數據流當作一個會不斷增長的表。這是一個新的流處理模型，但是與批處理模型很相似。你在做流式計算就像是標準針對靜態表的批查詢，Spark會在一個無限輸入的表上進行增量查詢。我們來從更多詳細內容來理解這個模型。

基本概念

將輸入的數據流理解為「寫入表」，每個流中到達的數據就像是寫入表中新增的一行。

針對輸入的查詢會生成「結果表」。每個觸發間隔之間（比如1秒鐘），就會有新的行添加到「寫入表」，最終更新結果表。當結果表變更後，我們能夠將變更的結果行寫入外部存儲。

「輸出（Output）」定義為寫入外部存儲的內容。輸出存在幾種模式：

• 完全模式（Complete Mode） ：整個更新後的結果表會全部寫入外部存儲。具體的全表寫入方式取決於與存儲的底層連接。
• 增量模式（Append Mode） ：從上次觸發後的新增結果表數據才會寫入外部存儲。這個模式只適用於那些預期結果表中的存量數據不會變化的查詢。
• 更新模式（Update Mode） ： 從上次觸發後的更新結果表數據才會寫入外部存儲（從Spark 2.1.1開始生效）。注意本模式和完全模式的差異，本模式下只會輸出上次觸發後的變更行。如果查詢不包含聚合，基本會和增量模式相同。

要注意每個模式都有確定的適配的查詢，這個會在稍後討論。
為了解釋這個模型的使用方式，我們用上面的快速示例來輔助理解模型。第一個DataFrame類型的變數 line 就是寫入表，而最後DataFrame類型的變數 wordCounts 就是結果表。注意針對流的查詢方法，從 line 生成 wordCounts 和一個靜態的DataFrame完全相同。當查詢開始之後，Spark會持續檢查從socket鏈接傳入的新數據。如果存在新數據，Spark會運行「增量」查詢，並且針對新數據計算更新的計數，整合之前運行的計數，如下圖所示。

注意結構化流並沒有存儲整張表。從數據源讀取最近有效的數據，增量的處理並且更新結果數據，然後丟棄源數據。Spark只保留最小中間狀態數據，用於更新結果（例如上面例子中的中間統計結果計數）。
這個模型明顯和其他的流處理引擎不同。許多流處理系統要求用戶自行維護運行聚合，因為有諸如容錯性（fault-tolerance）、數據一致性（data consistency：at-least-once, at-most-once, exactly-once）。在這個模型中，當有新數據時，由Spark負責更新結果表，因此解放了用戶無需關注。我們以模型處理事件時間和延遲數據作為例子來看下。

處理事件時間和延遲數據

事件時間是包含在數據本身的。很多應用都希望基於事件時間操作。例如你的想要獲取物聯網設備每分鐘產生事件數量，然後你可能希望使用數據生成的時間（這就是事件時間），而不是Spark接收到他們的時間。事件時間在這個模型中是很自然的，因為每個設備產生事件都是都是表中的一行數據，而事件時間就是一行數據中的一列。這樣基於窗口的聚合（例如每分鐘的事件數量）可以作為基於事件時間列做的特別的分組和聚合。每個時間窗口都是一個分組，每行數據也可以屬於多個窗口或分組。因此類似這種基於事件時間的聚合查詢能夠在靜態數據集（例如收集的設備事件日誌）和動態數據流，能夠是用戶的使用比較簡單。
此外模型天然的能夠基於事件時間處理延遲到達的數據。當Spark更新結果表時，他仍然能夠針對延遲數據來更新歷史聚合的結果，也同時可以清除歷史聚合數據，從而限制中間狀態數據的大小。從Spark2.1開始，我們支援水位線概念（watermarking），允許用戶指定延遲數據的閾值，系統也能夠清理舊狀態數據。稍後會在窗口操作章節介紹。

容錯性

保證唯一投送端到端是結構化流的設計中的關鍵目標之一。為了達成這樣的目標，我們設計了結構化流的源（Source）、匯（Sink）以及執行引擎能夠可靠的跟蹤處理進度，從而能夠重啟/重新處理來應對各種故障。每個數據流的源應該都有偏移量概念（類似Kafka的偏移量，或者Amazon Kinesis 的序列編號）來跟蹤流中的讀取位置。引擎使用保存點和先寫日誌來記錄每次處理的數據偏移邊界。流的匯設計天然就支援重新處理的冪等性。整合起來，使用可重放的源與冪等的匯，結構化流在面對任何故障時都能保證端對端嚴格一致性（end-to-end exactly-once semantics）。