Skip to content

12. 流處理

有效的複雜系統總是從簡單的系統演化而來。反之亦然：從零設計的複雜系統沒一個能有效工作的。

—— 約翰・加爾，Systemantics（1975）

在 第 11 章 中，我們討論了批處理技術：它以一組檔案作為輸入，並生成一組新的輸出檔案。輸出是 衍生資料（derived data） 的一種形式；也就是說，如有必要，可以再次執行批處理來重新建立這份資料集。我們看到，這個簡單而強大的思想可以用來構建搜尋索引、推薦系統和分析系統，等等。

然而，第 11 章 始終建立在一個重要假設之上：輸入是有界的，也就是大小已知且有限，因此批處理知道何時已經讀完輸入。例如，作為 MapReduce 核心環節的排序操作必須先讀完全部輸入，才能開始生成輸出。因為最後一條輸入記錄可能恰好擁有最小的鍵，因而需要成為第一條輸出記錄，所以不能提前開始輸出。

實際上，許多資料都是 無界的，因為它們會隨時間推移陸續到達：使用者昨天和今天產生了資料，明天還會繼續產生更多資料。只要你的企業還在經營，這個過程就不會結束，因此從任何有意義的角度來看，資料集都永遠不會“完整”1。所以，批處理程式不得不人為地按固定時長切分資料，例如每天結束時處理當天的資料，或每小時結束時處理這一小時的資料。

每日批處理的問題在於，輸入中的變化要到一天後才會反映到輸出中，對許多沒有耐心的使用者來說實在太慢。為了縮短延遲，可以更頻繁地執行處理——比如每秒結束時處理這一秒的資料——也可以完全拋開固定的時間切片，改為連續處理，每個事件一發生就立即處理。這就是 流處理（stream processing） 的基本思想。

一般來說，“流”是指隨時間推移逐漸可用的資料。這個概念出現在許多地方：Unix 的 stdin 和 stdout、程式語言中的惰性列表2、檔案系統 API（例如 Java 的 FileInputStream）、TCP 連線、透過網際網路傳輸的音訊和影片，等等。

本章將把 事件流（event stream） 作為一種資料管理機制來考察：它是上一章批次資料的無界、增量處理版本。我們首先討論如何表示、儲存流，以及如何透過網路傳輸流；接著在“資料庫與流”中研究流與資料庫的關係；最後在“流處理”中探討持續處理這些流的方法和工具，以及如何用它們構建應用。

傳遞事件流

在批處理領域，作業的輸入和輸出都是檔案（可能位於分散式檔案系統上）。那麼，流處理領域中的對應物是什麼？

當輸入是檔案（即位元組序列）時，第一個處理步驟通常是把它解析成一系列記錄。在流處理的語境中，記錄更常被稱為 事件（event），但兩者本質上是同一種東西：一個小巧、自包含、不可變的物件，記錄某個時刻發生的某件事。事件通常帶有時間戳，表示按照日曆時鐘來看它發生在何時（參見“單調時鐘與日曆時鐘”）。

例如，事件可能來自使用者的某個操作，如瀏覽頁面或完成購買；也可能來自機器，如溫度感測器的定期測量值或 CPU 利用率指標。在“使用 Unix 工具的批處理”示例中，Web 伺服器日誌的每一行就是一個事件。

事件可以編碼成文字字串、JSON，或 第 5 章 討論的某種二進位制格式。編碼後，你就可以儲存事件，例如把它追加到檔案、插入關係表，或寫入文件資料庫；也可以透過網路把事件傳送到另一個節點進行處理。

在批處理中，檔案寫入一次，之後可能由多個作業讀取。與之類似，在流處理術語中，事件由 生產者（producer）（也稱為 釋出者（publisher） 或 傳送者（sender））生成一次，之後可能由多個 消費者（consumer）（也稱為 訂閱者（subscriber） 或 接收者（recipient））處理3。在檔案系統中，檔名標識一組相關記錄；在流式系統中，相關事件通常被歸入同一個 主題（topic） 或 流（stream）。

原則上，檔案或資料庫足以把生產者和消費者連線起來：生產者將生成的每個事件寫入資料儲存，而各個消費者定期輪詢資料儲存，檢查自上次執行以來出現了哪些新事件。這實質上就是每日結束時處理當天資料的批處理程式所做的事情。

然而，當我們轉向低延遲的持續處理時，如果資料儲存並非為這種用法而設計，輪詢的代價就會很高。輪詢越頻繁，返回新事件的請求比例越低，額外開銷也就越大。更好的做法是在出現新事件時通知消費者。

傳統資料庫對這種通知機制的支援並不好。關係資料庫通常提供 觸發器（trigger），可以對變更作出響應（例如向表中插入一行），但觸發器能做的事情非常有限，在資料庫設計中多少有些事後補充的意味4。因此，人們開發了專門用於傳遞事件通知的工具。

訊息傳遞系統

向消費者通知新事件的一種常見方法是使用 訊息傳遞系統（messaging system）：生產者傳送包含事件的訊息，再由系統將訊息推送給消費者。我們之前在“事件驅動的架構”中提到過這類系統，現在來進一步瞭解其細節。

在生產者與消費者之間建立 Unix 管道或 TCP 連線等直接通訊通道，是實現訊息傳遞系統的一種簡單方法。不過，大多數訊息傳遞系統都擴充套件了這個基本模型。Unix 管道和 TCP 恰好連線一個傳送者與一個接收者，而訊息傳遞系統允許多個生產者節點向同一主題傳送訊息，也允許多個消費者節點接收一個主題中的訊息。

在這種 釋出/訂閱 模型中，不同系統採取了五花八門的方法，沒有一種答案適合所有用途。要區分這些系統，下面兩個問題尤其有用：

  1. 如果生產者傳送訊息的速度超過消費者的處理速度，會發生什麼？ 大體上有三種選擇：系統可以丟棄訊息、把訊息快取在佇列中，或施加 背壓（backpressure）（也稱為 流量控制（flow control），即阻塞生產者，令其暫停傳送更多訊息）。例如，Unix 管道和 TCP 都採用背壓：它們有一個固定大小的小緩衝區；如果緩衝區已滿，傳送者就會被阻塞，直到接收者從中取走資料（參見“網路擁塞與排隊”）。

    如果訊息被快取在佇列中，就必須弄清楚佇列不斷增長會有什麼後果：佇列大到記憶體容納不下時，系統會崩潰，還是會把訊息寫入磁碟？如果寫入磁碟，磁碟訪問會怎樣影響訊息傳遞系統的效能5？磁碟寫滿時又會發生什麼6？

  2. 如果節點崩潰或暫時離線，會發生什麼——是否會丟失訊息？ 和資料庫一樣，要實現永續性，可能需要以某種方式組合使用磁碟寫入與複製（參見側欄“複製與永續性”），而這會付出代價。如果能夠容忍偶爾丟失訊息，那麼在同樣的硬體上，通常可以獲得更高的吞吐量和更低的延遲。

能否容忍訊息丟失，很大程度上取決於應用。例如，對於定期傳送的感測器讀數和指標，偶爾缺失一個數據點或許並不重要，因為不久後還會發來更新值。不過要小心：如果大量訊息被丟棄，你可能無法立即察覺指標已經失真7。如果你要對事件計數，可靠傳遞就重要得多，因為每丟失一條訊息，計數器就會出現一份誤差。

我們在 第 11 章 探討的批處理系統有一個很好的特性：它們提供了很強的可靠性保證。失敗的任務會自動重試，失敗任務產生的部分輸出會自動丟棄。因此，最終輸出與從未發生故障時相同，這簡化了程式設計模型。本章後面將考察如何在流處理環境中提供類似的保證。

直接從生產者傳遞給消費者

許多訊息傳遞系統讓生產者與消費者直接透過網路通訊，不經過任何中間節點：

  • UDP 組播廣泛用於金融行業中的股票行情等資料流，因為這些場景十分看重低延遲8。儘管 UDP 本身並不可靠，但應用層協議可以恢復丟失的資料包（生產者必須記住已傳送的資料包，以便按需重傳）。

  • ZeroMQ 和 nanomsg 等無代理訊息庫採用類似的方法，透過 TCP 或 IP 組播實現釋出/訂閱訊息傳遞。

  • StatsD 等指標收集代理9使用不可靠的 UDP 訊息，從網路中的所有機器收集指標並進行監控。（在 StatsD 協議中，只有收到全部訊息，計數器指標才是準確的；使用 UDP 意味著這些指標至多隻能是近似值10。另見“TCP 與 UDP”。）

  • 如果消費者在網路上公開了一項服務，生產者可以直接發起 HTTP 或 RPC 請求（參見“流經服務的資料流：REST 與 RPC”），把訊息推送給消費者。Webhook11就建立在這一思想之上：把一個服務的回撥 URL 註冊到另一個服務中，後者每逢事件發生便向該 URL 發出請求。

儘管這些直接訊息傳遞系統在各自的目標場景中表現良好，但應用程式碼通常必須自行考慮訊息丟失的可能性。它們能容忍的故障十分有限：即使協議可以檢測並重傳網路中丟失的資料包，通常仍會假設生產者和消費者始終線上。

消費者離線時，可能會錯過其不可達期間傳送的訊息。有些協議允許生產者重試失敗的訊息傳遞，但如果生產者崩潰，丟失了本應重試的訊息緩衝區，這種辦法就可能失效。

訊息代理

一種廣泛使用的替代方案是透過 訊息代理（message broker）（也稱為 訊息佇列（message queue））傳送訊息。訊息代理實質上是一類針對訊息流最佳化的資料庫12。它作為伺服器執行，生產者和消費者則作為客戶端連線到它。生產者把訊息寫入代理，消費者透過讀取代理來接收訊息。

資料集中到代理之後，系統更容易容忍客戶端時而連線、時而斷開乃至崩潰，永續性問題也轉移給了代理。有些訊息代理只把訊息儲存在記憶體中，另一些則會根據配置把訊息寫入磁碟，以免代理崩潰時丟失訊息。面對緩慢的消費者，它們通常允許佇列無限增長，而不是丟棄訊息或施加背壓，不過具體行為也可能取決於配置。

排隊還有一個後果：消費者通常是 非同步的。生產者傳送訊息時，一般只等待代理確認訊息已被快取，並不等待消費者完成處理。訊息會在未來某個無法確定的時間點傳遞給消費者——往往不到一秒，但如果佇列中積壓了大量訊息，也可能晚得多。

訊息代理與資料庫的對比

有些訊息代理甚至可以透過 XA 或 JTA 參與兩階段提交協議（參見“跨不同系統的分散式事務”）。這一功能使它們在性質上與資料庫十分相似，不過訊息代理與資料庫之間仍有一些重要的實際差異：

  • 資料庫通常會一直儲存資料，直到有人顯式刪除；而有些訊息代理會在訊息成功傳遞給消費者後自動將其刪除。這類訊息代理不適合長期儲存資料。

  • 正因為訊息會很快刪除，大多數訊息代理假定工作集相當小，也就是佇列很短。如果消費者緩慢，導致代理必須緩衝大量訊息（記憶體容納不下時還可能溢寫到磁碟），每條訊息的處理時間就會延長，總體吞吐量也可能下降5。

  • 資料庫通常支援二級索引，並能透過查詢語言以多種方式搜尋資料；訊息代理通常只支援訂閱與某種模式匹配的一組主題。兩者本質上都讓客戶端選擇自己關心的資料子集，但資料庫提供的查詢功能通常強大得多。

  • 查詢資料庫時，結果通常以某一時刻的資料快照為依據；如果另一個客戶端隨後寫入資料庫並改變了查詢結果，第一個客戶端不會知道先前的結果已經過時，除非再次執行查詢或輪詢變更。相比之下，訊息代理不支援任意查詢，也不允許修改已經發出的訊息，但會在資料發生變化時（即有新訊息可用時）通知客戶端。

這是訊息代理的傳統形態，JMS13、AMQP14 等標準對它作出了規範，RabbitMQ、ActiveMQ、HornetQ、Qpid、TIBCO Enterprise Message Service、IBM MQ、Azure Service Bus 和 Google Cloud Pub/Sub 等軟體則實現了這種形態15。儘管也可以把資料庫用作佇列，但要把效能調到理想水平並不容易16。

多個消費者

多個消費者讀取同一主題中的訊息時，主要有兩種訊息傳遞模式，如 圖 12-1 所示：

負載均衡

每條訊息只傳遞給 一個 消費者，因此多個消費者可以分擔處理該主題訊息的工作。代理可以任意把訊息分配給消費者。如果訊息處理成本很高，而你希望透過增加消費者來並行處理，便適合採用這種模式。（在 AMQP 中，可以讓多個客戶端消費同一個佇列來實現負載均衡；在 JMS 中，這稱為 共享訂閱（shared subscription）。）

扇出

每條訊息都傳遞給 所有 消費者。扇出讓幾個相互獨立的消費者都能“收聽”同一份訊息廣播，彼此互不影響——相當於流處理版本的“幾個不同批處理作業讀取同一個輸入檔案”。（JMS 的主題訂閱和 AMQP 的交換器繫結提供了這一功能。）

圖 12-1. （a）負載均衡：由多個消費者分擔一個主題的消費工作；（b）扇出：將每條訊息傳遞給多個消費者。

這兩種模式可以結合使用，例如 Kafka 的 消費者組（consumer group） 功能。消費者組訂閱某個主題後，該主題中的每條訊息都會發送給組內的一個消費者（在組內消費者之間進行負載均衡）。如果兩個不同的消費者組訂閱同一主題，那麼每條訊息都會發送給各組中的一個消費者（在消費者組之間實現扇出）。

確認應答與重新傳遞

消費者隨時可能崩潰，因此代理可能已經把訊息傳遞給消費者，但消費者尚未處理，或只處理了一部分便發生崩潰。為了確保訊息不會丟失，訊息代理採用 確認應答（acknowledgment）：客戶端處理完訊息後，必須明確告知代理，代理才能將訊息從佇列中移除。

如果客戶端連線關閉或超時，而代理尚未收到確認應答，代理便假定訊息沒有得到處理，並把它重新傳遞給另一個消費者。（請注意，訊息可能 實際上已經 處理完畢，只是確認應答在網路中丟失了。除非該操作具有冪等性，或不要求恰好一次語義，否則必須使用原子提交協議來處理這種情況，詳見“恰好一次訊息處理”。）

重新傳遞與負載均衡結合後，會對訊息順序產生一個有趣的影響。在 圖 12-2 中，消費者通常按照生產者傳送訊息的順序進行處理。但是，消費者 2 在處理訊息 m3 時崩潰，與此同時消費者 1 正在處理訊息 m4。尚未確認的訊息 m3 隨後被重新傳遞給消費者 1，於是消費者 1 按照 m4、m3、m5 的順序處理訊息。因此，m3 和 m4 的傳遞順序與生產者 1 的傳送順序不同。

圖 12-2. 消費者 2 在處理 m3 時崩潰，因此 m3 稍後被重新傳遞給消費者 1。

即使訊息代理試圖保持訊息順序（JMS 和 AMQP 標準都作此要求），負載均衡與重新傳遞的結合仍不可避免地會使訊息發生重排。要避免這個問題，可以為每個消費者使用單獨的佇列，也就是不使用負載均衡。如果訊息彼此完全獨立，重排並無大礙；但如果訊息之間存在因果依賴，順序就可能十分重要，本章後面將看到這一點。

重新傳遞還可能浪費資源、造成資源飢餓，甚至永久阻塞一條流。常見情形是生產者沒有正確序列化訊息，例如編碼成 JSON 的物件缺少必填鍵。任何讀到這條訊息的消費者都會期待該鍵存在，並在發現缺失時失敗。由於沒有發出確認應答，代理會再次傳送訊息，使另一個消費者也隨之失敗，如此無限迴圈。如果代理提供嚴格的順序保證，後續處理將完全無法推進。允許訊息重排的代理仍能繼續處理其他訊息，卻會把資源浪費在永遠得不到確認的訊息上。

死信佇列（dead letter queue，DLQ） 用來處理這類問題。系統不再把訊息留在當前佇列中無限重試，而是將其移到另一條佇列，讓消費者能夠繼續前進17、18。通常會對死信佇列設定監控——佇列中出現任何訊息都意味著發生了錯誤。檢測到新訊息後，操作員可以決定永久丟棄它、手動修改後重新生產這條訊息，或修復消費者程式碼，使其能夠正確處理訊息。大多數佇列系統都提供 DLQ；如今，Apache Pulsar 等基於日誌的訊息系統以及 Kafka Streams 等流處理系統也開始支援它19。

基於日誌的訊息代理

透過網路傳送資料包或向網路服務發出請求，通常都是轉瞬即逝的操作，不會留下永久痕跡。儘管可以透過抓包和日誌記錄把這些操作永久儲存下來，但我們通常不會這樣看待它們。AMQP/JMS 風格的訊息代理繼承了這種臨時訊息傳遞的思路：即使把訊息寫入磁碟，也會在訊息傳遞給消費者後很快將其刪除。

資料庫和檔案系統的思路恰恰相反：凡是寫入資料庫或檔案的內容，通常都應該永久儲存，至少要儲存到有人明確決定再次刪除它為止。

這種思路上的差異會極大地影響衍生資料的建立方式。正如 第 11 章 所說，批處理的一項關鍵特性是可以反覆執行，試驗不同的處理步驟，又不用擔心損壞輸入，因為輸入是隻讀的。AMQP/JMS 風格的訊息傳遞卻並非如此：如果確認應答會讓代理刪除訊息，那麼接收訊息就是一種破壞性操作。你無法重新運行同一個消費者，並期望得到同樣的結果。

向訊息傳遞系統加入新消費者時，它通常只能接收註冊之後發出的訊息；先前的訊息早已消失，無法恢復。相比之下，檔案和資料庫可以隨時加入新客戶端，並讀取任意久遠的資料，只要應用沒有明確覆蓋或刪除這些資料。

為什麼不能把兩者結合起來，既採用資料庫的持久儲存方式，又具備訊息傳遞的低延遲通知能力？這就是 基於日誌的訊息代理（log-based message broker） 的思想。近年來，這類系統已經十分流行。

使用日誌進行訊息儲存

日誌就是磁碟上一系列僅追加的記錄。我們此前在 第 4 章 討論日誌結構儲存引擎和預寫日誌時、在 第 6 章 討論複製時，以及在 第 10 章 把日誌作為一種共識形式來討論時，都見過這種結構。

同一種結構也可以用來實現訊息代理：生產者把訊息追加到日誌末尾，以此傳送訊息；消費者順序讀取日誌，以此接收訊息。消費者讀到日誌末尾後，便等待新訊息追加的通知。用於監視檔案新增內容的 Unix 工具 tail -f，實質上的工作方式與此相同。

為了把吞吐量擴充套件到單塊磁碟的能力之上，可以對日誌進行 分片（即 第 7 章 所說的分片）。不同分片可以託管在不同機器上，每個分片都是一份獨立於其他分片讀寫的日誌。一個主題可以定義為一組承載相同型別訊息的分片。圖 12-3 展示了這種方法。

在每個分片中——Kafka 把它稱為 分割槽（partition）——代理會為每條訊息分配一個單調遞增的序列號，也就是 偏移量（offset）（圖 12-3 方框中的數字就是訊息偏移量）。分割槽是僅追加的，因此這樣的序列號有明確含義：分割槽內的訊息具有全序，而不同分割槽之間沒有順序保證。

圖 12-3. 生產者把訊息追加到主題分割槽檔案中，消費者則順序讀取這些檔案。

Apache Kafka20 和 Amazon Kinesis Streams 都是以這種方式工作的基於日誌的訊息代理。Google Cloud Pub/Sub 的架構與之相似，但公開的是 JMS 風格的 API，而不是日誌抽象15。儘管這些訊息代理會把所有訊息寫入磁碟，但透過跨多臺機器分片，仍能達到每秒數百萬條訊息的吞吐量，並透過複製訊息實現容錯21、22。

日誌與傳統訊息傳遞的比較

基於日誌的方法自然支援扇出式訊息傳遞，因為多個消費者可以各自讀取日誌而互不影響——讀取訊息不會把它從日誌中刪除。要在一組消費者之間實現負載均衡，代理可以把整個分片分配給消費者組中的節點，而不是把一條條訊息分別分配給各個消費者客戶端。

隨後，每個客戶端會消費所分配分片中的 全部 訊息。消費者得到一個日誌分片後，通常會採用直截了當的單執行緒方式，順序讀取其中的訊息。這種粗粒度負載均衡有一些缺點：

  • 分擔主題消費工作的節點數，最多隻能等於該主題的日誌分片數，因為同一分片內的訊息都傳遞給同一個節點。（也可以設計一種讓兩個消費者共同處理一個分片的負載均衡方案：兩者都讀取完整訊息集，但一個只處理偏移量為偶數的訊息，另一個只處理偏移量為奇數的訊息。另一種辦法是把訊息處理分派給執行緒池，但這會使消費者偏移量的管理更加複雜。一般來說，最好還是用單執行緒處理一個分片，並透過增加分片來提高並行度。）

  • 如果某一條訊息處理得很慢，就會阻塞該分片中後續訊息的處理，這是一種隊頭阻塞（參見“描述效能”）。

因此，如果訊息處理成本很高，你希望以訊息為單位並行處理，而且訊息順序並不十分重要，那麼 JMS/AMQP 風格的訊息代理更合適。反之，如果訊息吞吐量很高，每條訊息都能迅速處理，而且順序非常重要，基於日誌的方法就表現出色23、24。不過，兩種架構之間的界線正在變得模糊：Kafka 等基於日誌的訊息系統如今也支援 JMS/AMQP 風格的消費者組，讓多個消費者能夠接收同一分割槽中的訊息25、26。

由於分片日誌通常只能保持單個分片內部的訊息順序，所有必須以一致順序處理的訊息都需要路由到同一個分片。例如，應用可能要求與某個特定使用者有關的事件始終以固定順序出現。可以根據事件的使用者 ID 選擇分片來實現這一點；換句話說，把使用者 ID 作為 分割槽鍵（partition key）。

消費者偏移量

順序消費一個分片，很容易判斷哪些訊息已經處理：偏移量小於消費者當前偏移量的訊息都已處理，偏移量更大的訊息則尚未讀到。因此，代理無需跟蹤每條訊息的確認應答，只需定期記錄消費者偏移量。這樣既減少了簿記開銷，也帶來了批次處理和流水線化的機會，有助於提高基於日誌系統的吞吐量。不過，如果消費者發生故障，它會從上次記錄的偏移量恢復，而不是從自己實際讀到的最新位置恢復，因此可能會再次讀到一些訊息。

這種偏移量其實與單主資料庫複製中常見的 日誌序列號 非常相似，我們曾在“設定新的副本”中討論過它。在資料庫複製中，日誌序列號允許追隨者斷開連線後重新連線到領導者，並且不跳過任何寫入就恢復複製。這裡採用的原理完全相同：訊息代理扮演領導者資料庫的角色，消費者則像追隨者。

如果消費者節點發生故障，消費者組會把它的分片分配給另一個節點，後者從最後記錄的偏移量開始消費。如果原消費者已經處理了後續訊息，卻尚未記錄相應的偏移量，那麼重啟後這些訊息會被再次處理。本章後面將討論如何應對這一問題。

磁碟空間使用

如果永遠只向日志追加內容，磁碟空間終有耗盡之時。為了回收空間，日誌實際上會被切成若干段，舊段會不時被刪除或移入歸檔儲存。（我們將在“日誌壓實”中討論一種更複雜的空間回收方法。）

這意味著，如果緩慢的消費者跟不上訊息產生的速度，落後到其偏移量指向已經刪除的日誌段，就會錯過一些訊息。實際上，日誌實現了一個大小有限的緩衝區，填滿後便丟棄舊訊息，也就是 迴圈緩衝區（circular buffer） 或 環形緩衝區（ring buffer）。不過，由於這個緩衝區位於磁碟上，它可以相當大。

讓我們粗略估算一下。在本書寫作時，典型的大容量硬碟為 20 TB，順序寫入吞吐量為 250 MB/s。如果始終以最高速度寫入訊息，大約 22 小時後磁碟就會寫滿，必須開始刪除最舊的訊息。這意味著，即使有許多機器和許多磁碟，磁碟日誌也總能緩衝至少 22 小時的訊息，因為增加磁碟會同時增加可用空間和總寫入頻寬。實際部署很少用滿磁碟的全部寫入頻寬，所以日誌通常可以儲存數天乃至數週的訊息。

許多基於日誌的訊息代理如今會把訊息存入物件儲存，以擴大儲存容量，這與“以物件儲存為後端的資料庫”中討論的做法相似。Apache Kafka 和 Redpanda 等訊息代理透過分層儲存從物件儲存提供較舊的訊息；WarpStream、Confluent Freight 和 Bufstream 等系統則把全部資料都存入物件儲存。除了成本效益，這種架構也簡化了資料整合：物件儲存中的訊息以 Iceberg 表儲存，批處理作業和資料倉庫作業可以直接在這些資料上執行，無需先把資料複製到另一個系統。

當消費者跟不上生產者時

在“訊息傳遞系統”開頭，我們討論了消費者跟不上生產者傳送訊息的速度時可採取的三種辦法：丟棄訊息、緩衝訊息或施加背壓。按照這種分類，基於日誌的方法屬於緩衝，它提供了一個很大但大小固定的緩衝區，其上限由可用磁碟空間決定。

如果消費者遠遠落後，需要的訊息已經早於磁碟保留範圍，它就無法再讀取這些訊息——也就是說，代理實際上會丟棄超出緩衝容量的舊訊息。你可以監控消費者落後日誌頭部多遠，並在落後過多時發出告警。由於緩衝區很大，通常有足夠時間讓運維人員修復緩慢的消費者，並在它開始漏掉訊息前追上進度。

即使某個消費者確實落後太多並開始漏掉訊息，也只有它自己受到影響，不會干擾其他消費者的服務。這是一項很大的運維優勢：你可以出於開發、測試或除錯目的，試驗性地消費生產日誌，不必太擔心干擾生產服務。消費者關閉或崩潰後便不再消耗資源，唯一留下的只是它的偏移量。

這種行為也與傳統訊息代理形成鮮明對比。在傳統代理中，必須小心刪除消費者已經關閉的佇列，否則這些佇列會繼續積累不再需要的訊息，擠佔仍在活動的消費者可用的記憶體。

重播舊訊息

前面提到，對於 AMQP/JMS 風格的訊息代理，處理並確認訊息是一種破壞性操作，因為這會使代理刪除訊息。而在基於日誌的訊息代理中，消費訊息更像讀取檔案：它是一種不會改變日誌的只讀操作。

除了消費者本身產生的輸出，處理訊息唯一的副作用就是消費者偏移量向前移動。但偏移量由消費者控制，所以必要時很容易調整。例如，可以用昨天的偏移量啟動一份消費者副本，把輸出寫到另一個位置，從而重新處理過去一天的訊息。你可以任意重複這個過程，每次換用不同的處理程式碼。

這一特性讓基於日誌的訊息傳遞更像上一章的批處理：透過可重複執行的轉換過程，把衍生資料與輸入資料明確分開。它為試驗提供了更多空間，也更容易從錯誤和程式缺陷中恢復，因此很適合用來整合組織內部的資料流27。

資料庫與流

前面我們對訊息代理與資料庫做過一些比較。傳統上，它們被視為兩類不同的工具，但基於日誌的訊息代理已經成功地把資料庫中的思想用於訊息傳遞。反過來也一樣：我們可以把訊息傳遞和流中的思想用於資料庫。

一種做法是用 事件流充當儲存資料的權威記錄系統（參閱 “權威記錄系統與衍生資料”）。這正是我們在 “事件溯源與 CQRS” 中討論過的 事件溯源：不再用更新和刪除來改變資料模型，而是把每次狀態變化建模成不可變事件，寫入僅追加日誌；所有為讀取最佳化的物化檢視都從這些事件中衍生出來。基於日誌的訊息代理採用僅追加儲存，還能以低延遲通知消費者有新事件到來，因此很適合事件溯源——只要將其配置為永不刪除舊事件。

不過，你不必走到採用事件溯源這一步；即便資料模型是可變的，事件流對資料庫仍然很有用。事實上，每次資料庫寫入都是一個可以捕獲、儲存和處理的事件。資料庫與流的聯絡不只是日誌在磁碟上的物理儲存形式，而是更為根本。

例如，複製日誌（參閱 “複製日誌的實現”）就是資料庫寫入事件組成的流，由領導者在處理事務時生成。追隨者把這股寫入流應用到自己的資料庫副本上，最終得到同一份資料的準確副本。複製日誌中的事件描述的正是已經發生的資料變化。

我們還在 “使用共享日誌” 中遇到過 狀態機複製 原理：如果每個事件都代表一次資料庫寫入，並且每個副本都以相同順序處理相同事件，那麼所有副本最終都會達到相同狀態（這裡假定事件處理是確定性的操作）。這又是事件流的一個例子！

本節先考察異構資料系統中會出現的一個問題，再探討如何把事件流的思想引入資料庫來解決它。

保持系統同步

正如全書反覆說明的，沒有一個系統能夠滿足所有資料儲存、查詢和處理需求。實踐中，大多數稍具規模的應用都要組合多種技術才能滿足要求：例如用 OLTP 資料庫處理使用者請求，用快取加速常見請求，用全文索引處理搜尋查詢，再用資料倉庫進行分析。每個系統都有自己的資料副本，並採用為自身用途最佳化的表示形式。

相同或相關的資料分散在不同位置，就必須彼此保持同步：資料庫中的某個專案更新後，快取、搜尋索引和資料倉庫也要隨之更新。資料倉庫通常透過 ETL 流程完成同步（參閱 “資料倉庫”）：先取得資料庫的完整副本，轉換資料，再批次載入進資料倉庫——換言之，這是一個批處理過程。同樣，我們在 “批處理用例” 中看到，搜尋索引、推薦系統以及其他衍生資料系統也可以透過批處理來建立。

如果定期轉儲完整資料庫太慢，有時會改用 雙寫：資料發生變化時，應用程式碼顯式寫入每個系統，例如先寫資料庫，再更新搜尋索引，最後使相應的快取項失效（也可能併發執行這些寫入）。

然而，雙寫存在一些嚴重問題，其中之一就是 圖 12-4 所示的競態條件。在這個例子中，兩個客戶端併發更新專案 X：客戶端 1 想把值設為 A，客戶端 2 想把值設為 B。兩個客戶端都先把新值寫入資料庫，再寫入搜尋索引。由於時序不巧，請求交錯執行：資料庫先收到客戶端 1 將值設為 A 的寫入，再收到客戶端 2 將值設為 B 的寫入，因此資料庫中的最終值為 B；搜尋索引卻先收到客戶端 2 的寫入，再收到客戶端 1 的寫入，因此最終值為 A。雖然沒有發生任何錯誤，兩個系統卻永久地不一致了。

圖 12-4. 資料庫中的 X 先被設為 A，再被設為 B；寫入到達搜尋索引的順序卻恰好相反。

除非另外採用併發檢測機制，例如我們在 “檢測併發寫入” 中討論的版本向量，否則你甚至不會察覺發生過併發寫入——一個值只會悄無聲息地覆蓋另一個值。

雙寫的另一個問題是，其中一次寫入可能失敗，另一次卻成功。這屬於容錯問題而不是併發問題，但同樣會使兩個系統彼此不一致。要保證兩次寫入要麼都成功，要麼都失敗，就要解決代價高昂的原子提交問題（參閱 “兩階段提交（2PC）”）。

如果只有一個採用單主複製的資料庫，那麼領導者會決定寫入順序，狀態機複製就能在資料庫的各個副本之間正常工作。然而，圖 12-4 中並不存在唯一的領導者：資料庫可能有自己的領導者，搜尋索引也可能有自己的領導者，但兩者誰也不追隨誰，因而可能發生衝突（參閱 “多主複製”）。

如果真能只有一個領導者——例如資料庫——並讓搜尋索引成為資料庫的追隨者，情況就會好得多。但實踐中能做到嗎？

變更資料捕獲

大多數資料庫的複製日誌長期以來都被視為內部實現細節，而不是公共 API。客戶端理應透過資料庫的資料模型和查詢語言進行查詢，而不是解析複製日誌，嘗試從中提取資料。

幾十年來，許多資料庫根本沒有提供文件化的方式來獲取寫入其中的變更日誌。因此，要取得資料庫中發生的所有變化，再將其複製到搜尋索引、快取或資料倉庫等其他儲存技術中，一直非常困難。

近年來，變更資料捕獲（change data capture，CDC）越來越受關注。它是這樣一個過程：觀察寫入資料庫的所有資料變化，將其提取成可以複製到其他系統的形式28。如果變化一經寫入就立即以流的形式提供出來，CDC 尤其有用。

例如，你可以捕獲資料庫中的變化，並持續把相同變化應用到搜尋索引。只要按同一順序應用變更日誌，搜尋索引中的資料就有望與資料庫保持一致。搜尋索引以及其他衍生資料系統，都只是變更流的消費者。

圖 12-5 展示了 CDC 如何解決 圖 12-4 中的併發問題。將 X 分別設為 A 和 B 的兩個請求雖然併發到達資料庫，但資料庫會決定某種執行順序，並按該順序把它們寫入複製日誌。搜尋索引再按相同順序取得並應用這些變化。如果還需要把資料送往資料倉庫等其他系統，只須再為 CDC 事件流新增一個消費者。

圖 12-5. 按資料寫入一個數據庫的順序取得資料，再按相同順序將變化應用到其他系統。

變更資料捕獲的實現

按照 “權威記錄系統與衍生資料” 中的說法，我們可以把日誌消費者稱為 衍生資料系統：搜尋索引和資料倉庫中儲存的資料，不過是權威記錄系統中資料的另一種檢視。變更資料捕獲機制確保權威記錄系統中的所有變化也會反映到衍生資料系統中，使衍生系統持有準確的資料副本。

實質上，變更資料捕獲讓一個數據庫成為領導者（即從中捕獲變化的資料庫），讓其他系統成為追隨者。基於日誌的訊息代理能夠保持訊息順序，避免 圖 12-2 中的亂序問題，因此很適合把變更事件從源資料庫傳送到衍生系統。

邏輯複製日誌可以用來實現變更資料捕獲（參閱 “邏輯（基於行）的日誌複製”），不過需要應對模式變更、恰當建模更新等挑戰。開源專案 Debezium 正是為解決這些問題而生。它為 MySQL、PostgreSQL、Oracle、SQL Server、Db2、Cassandra 以及其他許多資料庫提供了 源聯結器。這些聯結器接入資料庫複製日誌，以標準事件模式呈現其中的變化；隨後便可轉換訊息並將其寫入下游資料庫。Kafka Connect 框架也為各種資料庫提供了更多 CDC 聯結器。Maxwell 透過解析 binlog 為 MySQL 提供類似功能29；GoldenGate 為 Oracle 提供類似功能；pgcapture 則面向 PostgreSQL。

與訊息代理一樣，變更資料捕獲通常也是非同步的：權威記錄資料庫提交變化之前，並不會等待消費者應用該變化。這種設計在運維上的好處是，增加一個緩慢的消費者不會對權威記錄系統造成太大影響；缺點則是複製延遲的所有問題同樣存在（參閱 “複製延遲的問題”）。

初始快照

如果擁有資料庫有史以來的全部變更日誌，就可以透過重播日誌來重建資料庫的完整狀態。然而，永久保留所有變化往往會佔用太多磁碟空間，重播也會耗時過久，因此日誌通常需要截斷。

例如，構建新的全文索引需要整個資料庫的完整副本——只應用最近的變更日誌還不夠，因為其中缺少最近沒有更新過的專案。因此，如果沒有完整的歷史日誌，就需要從一個一致的快照開始，正如 “設定新的副本” 中所討論的那樣。

資料庫快照必須與變更日誌中的某個已知位置或偏移量對應，這樣才能知道快照處理完畢後應從何處開始應用變化。有些 CDC 工具集成了快照功能，有些則需要手工完成。Debezium 使用 Netflix 的 DBLog 水位線演算法提供增量快照30、31。

日誌壓實

如果只能保留有限的日誌歷史，那麼每增加一個新的衍生資料系統，都要重新執行一遍快照流程。不過，日誌壓實（log compaction）提供了一個很好的替代方案。

我們在 “日誌結構儲存” 中討論日誌結構儲存引擎時介紹過日誌壓實（示例見 圖 4-3）。原理很簡單：儲存引擎定期查詢日誌中鍵相同的記錄，丟棄重複項，只保留每個鍵的最新更新。這可能會顯著縮小日誌段，因此在壓實過程中也可以合併日誌段，如 圖 12-6 所示。整個過程在後臺執行。

圖 12-6. 一個鍵值對日誌：鍵是貓咪影片的 ID（mew、purr、scratch 或 yawn），值是播放次數。日誌壓實只保留每個鍵的最新值。

在日誌結構儲存引擎中，帶有特殊空值的更新（稱為 墓碑）表示某個鍵已被刪除，並使該鍵在日誌壓實時被移除。但只要一個鍵沒有被覆蓋或刪除，它就會永久留在日誌中。壓實後的日誌所需磁碟空間只取決於資料庫當前的內容，而與資料庫有史以來發生過多少次寫入無關。如果同一個鍵經常被覆蓋，舊值最終會被垃圾回收，只留下最新值。

同樣的思路也適用於基於日誌的訊息代理和變更資料捕獲。如果 CDC 系統保證每次變化都有主鍵，而且對某個鍵的每次更新都會取代該鍵的舊值，那麼只保留這個鍵最近一次寫入就足夠了。

這樣一來，每當需要重建搜尋索引等衍生資料系統時，都可以讓新消費者從經過日誌壓實的主題的偏移量 0 開始，依次掃描日誌中的所有訊息。日誌保證包含資料庫中每個鍵的最新值（也可能包含一些舊值）——換言之，無須再次對 CDC 源資料庫建立快照，便可由此取得資料庫內容的完整副本。

Apache Kafka 支援日誌壓實。正如本章後面將會看到的，它使訊息代理不僅可以傳送臨時訊息，還能用作持久儲存。

變更流的 API 支援

如今，大多數主流資料庫都把變更流作為一等介面公開出來，而不再依賴過去那種事後加裝、逆向工程得到的 CDC。MySQL、PostgreSQL 等關係資料庫通常透過自身副本所使用的同一份複製日誌傳送變化。大多數雲廠商也為自家產品提供 CDC 方案：例如，Datastream 可以流式訪問 Google Cloud 的關係資料庫和資料倉庫。

即便 Cassandra 這類最終一致、基於法定人數的資料庫，如今也支援變更資料捕獲。正如 “線性一致性與仲裁” 中所述，客戶端必須把寫入持久化到多數節點，該寫入才被視為可見。法定人數寫入很難支援 CDC，因為沒有唯一的權威資料來源可供訂閱；資料是否可見，取決於每個讀取者選擇的一致性級別。Cassandra 繞開了這個問題：它不提供統一的變更流，而是公開每個節點的原始日誌段。消費資料的系統必須讀取每個節點的原始日誌段，再自行決定如何將其合併成一股流，做法很像法定人數讀取者32。

Kafka Connect33 把許多資料庫系統的變更資料捕獲工具與 Kafka 整合起來。變更事件進入 Kafka 之後，既可以用來更新搜尋索引等衍生資料系統，也可以送入本章後面討論的流處理系統。

變更資料捕獲與事件溯源

我們來比較一下變更資料捕獲與事件溯源。與變更資料捕獲相似，事件溯源也把應用狀態的所有變化存成變更事件日誌。兩者最大的區別在於抽象層次不同：

  • 在變更資料捕獲中，應用以可變方式使用資料庫，可以隨意更新和刪除記錄。變更日誌從資料庫底層提取（例如解析複製日誌），從而確保提取出的寫入順序與實際寫入順序一致，避免 圖 12-4 中的競態條件。

  • 在事件溯源中，應用邏輯顯式構建在寫入事件日誌的不可變事件之上。事件儲存只允許追加，通常不鼓勵或禁止更新、刪除事件。事件旨在反映應用層發生的事情，而不是底層的狀態變化。

哪一種更好取決於具體情況。對於原本沒有采用事件溯源的應用，改用事件溯源是一項重大變化，也會帶來 “事件溯源與 CQRS” 中討論的各種利弊。相比之下，CDC 可以用很少的改動接入現有資料庫——寫入資料庫的應用甚至可能根本不知道 CDC 正在執行。

變更資料捕獲與資料庫模式

變更資料捕獲看起來比事件溯源更容易採用，但它也有自己的一系列挑戰。

在微服務架構中，一個數據庫通常只由一個服務訪問。其他服務透過該服務的公共 API 與之互動，一般不會直接訪問資料庫。這樣，資料庫就成為該服務的內部實現細節，開發者可以改變資料庫模式而不影響公共 API。

然而，CDC 系統複製資料時通常會沿用上游資料庫的模式，這會讓這些模式變成公共 API，必須像服務的公共 API 一樣加以管理。如果開發者刪除資料庫表中的一列，依賴該欄位的下游消費者就會崩潰。這類挑戰一直存在於資料流水線中，但過去通常只影響資料倉庫 ETL。CDC 往往以資料流實現，其他生產服務也可能是消費者，因此破壞這些消費者可能導致面向客戶的故障34。人們通常用資料契約來防止這類破壞。

將內部模式與外部模式解耦的一種常見方法是採用 發件箱模式（outbox pattern）。發件箱是具有獨立模式的表；CDC 系統對外公開這些表，而不是資料庫中的內部領域模型35、36。這樣，開發者便可按需修改內部模式，而保持發件箱表不變。這看起來像雙寫——它確實就是雙寫。不過，兩次寫入都留在同一個系統（資料庫）中，可以出現在同一個事務裡，因此發件箱避開了 “保持系統同步” 中討論的問題。

不過，發件箱也有一些權衡。開發者仍須維護內部模式與發件箱模式之間的轉換，這可能並不容易。發件箱還會增加資料庫寫入底層儲存的資料量，可能引發效能問題。

與變更資料捕獲一樣，重播事件日誌可以重建系統當前狀態。不過，兩者處理日誌壓實的方式不同：

  • 記錄更新的 CDC 事件通常包含記錄的完整新版本，因此一個主鍵的當前值完全由該主鍵的最新事件決定，日誌壓實可以丟棄同一主鍵的舊事件。

  • 事件溯源的建模層次更高：事件通常表達使用者操作的意圖，而不是該操作引發狀態更新的具體機制。後續事件通常不會覆蓋先前事件，因此需要完整的事件歷史才能重建最終狀態，不能用同樣的方式壓實日誌。

使用事件溯源的應用通常會儲存由事件日誌衍生出的當前狀態快照，以免反覆處理完整日誌。不過，這只是一種效能最佳化，用來加快讀取和崩潰恢復；系統的設計意圖仍是永久儲存所有原始事件，並能在需要時重新處理完整的事件日誌。我們將在 “不變性的侷限” 中討論這一假設。

狀態、流和不變性

我們在 第 11 章 中看到，批處理受益於輸入檔案的不變性：你可以在現有輸入檔案上執行實驗性的處理作業，而不必擔心損壞這些檔案。正是不變性原則賦予了事件溯源和變更資料捕獲如此強大的能力。

我們通常認為資料庫儲存著應用的當前狀態——這種表示針對讀取進行了最佳化，一般也最便於處理查詢。狀態的本質就在於它會變化，所以資料庫除了插入資料，還支援更新和刪除。這與不變性又如何相容？

只要狀態會變化，它就是一段時間內各種事件改變它的結果。例如，當前可用座位列表取決於已經處理過的預訂；當前賬戶餘額取決於賬戶的貸記與借記；Web 伺服器的響應時間圖，則是所有已發生 Web 請求各自響應時間的聚合。

無論狀態如何變化，總有一系列事件導致這些變化。事情可以做了再撤銷，但那些事件確實發生過這一事實不會改變。關鍵在於，可變狀態與不可變事件的僅追加日誌並不矛盾，而是一枚硬幣的兩面。所有變化構成的日誌——即 變更日誌（changelog）——表示了狀態如何隨時間演化。

如果你偏愛數學，可以說應用狀態是事件流對時間的積分，而變更流是狀態對時間的微分，如 圖 12-7 所示37、38。這個類比有其侷限（例如，狀態的二階導數似乎沒有什麼意義），但它是思考資料的一個實用起點。

圖 12-7. 應用當前狀態與事件流之間的關係。

只要持久儲存變更日誌，狀態便可以重現。如果把事件日誌視為權威記錄系統，把所有可變狀態都看作由它衍生而來，系統中的資料流就更容易推理。正如 Jim Gray 和 Andreas Reuter 在 1992 年所說39：

從根本上說，根本沒有必要保留資料庫；日誌已經包含了全部資訊。之所以要儲存資料庫（即日誌末尾所對應的當前狀態），只是為了提升檢索操作的效能。

日誌壓實是銜接日誌與資料庫狀態的一種方式：它只保留每條記錄的最新版本，丟棄已經被覆蓋的版本。

不可變事件的優點

資料庫中的不變性是一個古老的觀念。例如，會計師幾個世紀以來一直在財務記賬中運用不變性。一筆交易發生後，會被記入僅追加的 分類賬；分類賬本質上就是事件日誌，描述貨幣、商品或服務的轉手。損益表、資產負債表等賬目，則是彙總分類賬中的交易而衍生出來的40。

如果出了差錯，會計師不會刪除或修改分類賬中的錯誤交易，而是另加一筆交易來抵消錯誤，例如退還一筆誤收的費用。錯誤交易會永遠保留在分類賬中，因為它對審計可能十分重要。如果根據錯誤分類賬得出的錯誤數字已經公佈，下一個會計期間的數字就會包含相應的更正。這在會計工作中再正常不過41。

這種可審計性對金融系統尤其重要，但許多不受嚴格監管的系統也能從中受益。如果你不慎部署了一段有缺陷的程式碼，把錯誤資料寫進資料庫，而這段程式碼還能以破壞性的方式覆蓋資料，恢復起來會困難得多。有了不可變事件的僅追加日誌，診斷事情經過並從問題中恢復就容易得多。同樣，客服人員也可以利用審計日誌診斷客戶的請求和投訴。

不可變事件所包含的資訊也比當前狀態更加豐富。例如在購物網站上，顧客可能先把一件商品加入購物車，隨後又將其移除。從履行訂單的角度看，第二個事件抵消了第一個事件；但從分析角度看，知道顧客曾考慮購買某件商品、後來又放棄，也許很有用。或許他們日後會購買，或許他們找到了替代品。這些資訊會留在事件日誌中；如果資料庫在商品移出購物車時就刪除相應記錄，資訊也會隨之丟失。

從同一事件日誌中派生多個檢視

此外，把可變狀態與不可變事件日誌分離之後，還可以從同一份事件日誌衍生出幾種面向不同讀取方式的表示。這就像一股流有多個消費者一樣（圖 12-5）：例如，分析資料庫 Druid 會以這種方式直接從 Kafka 攝取資料，Kafka Connect 的匯聚聯結器則可以把 Kafka 中的資料匯出到各種資料庫和索引33。

在事件日誌與資料庫之間加入顯式的轉換步驟，應用也更容易隨時間演進。如果要引入一項新功能，用新的方式呈現現有資料，可以利用事件日誌為新功能構建一個獨立的、針對讀取最佳化的檢視，與現有系統並行執行，而不必修改現有系統。在許多情況下，新舊系統並行執行比在現有系統中執行複雜的模式遷移更加容易。等讀取方全部切換到新系統、不再需要舊系統之後，只須關閉舊系統並回收資源即可42、43。

把資料寫成一種針對寫入最佳化的形式，再按需轉換成多種針對讀取最佳化的表示，這正是我們在 “事件溯源與 CQRS” 中見過的 命令查詢職責分離（command query responsibility segregation，CQRS）模式。它不一定要求採用事件溯源：同樣可以從 CDC 事件流構建多個物化檢視44。

傳統的資料庫與模式設計方法建立在一個謬誤之上：資料必須按將來查詢它時所採用的形式寫入。如果能把針對寫入最佳化的事件日誌轉換成針對讀取最佳化的應用狀態，正規化與反正規化之爭（參閱 “正規化、反正規化與連線”）就基本失去了意義。完全可以在讀取最佳化檢視中對資料做反正規化，因為轉換過程提供了讓檢視與事件日誌保持一致的機制。

在 “案例研究：社交網路首頁時間線” 中，我們討論過社交網路的主頁時間線：它快取著某位使用者關注的人最近釋出的帖子，就像郵箱一樣。這也是針對讀取最佳化的狀態：主頁時間線高度反正規化，因為你的帖子會複製到每位關注者的時間線中。不過，扇出服務會讓這些重複狀態與新帖子、新的關注關係保持同步，使這種重複仍然可控。

併發控制

CQRS 最大的缺點是事件日誌的消費者通常以非同步方式執行。因此，使用者可能剛剛向日志寫入資料，隨即讀取某個衍生檢視，卻發現該寫入還沒有反映到檢視中。我們曾在 “讀己之寫” 中討論過這個問題及其可能的解決辦法。

一種解決辦法是在把事件追加到日誌時，同步更新讀取檢視。這要麼需要在事件日誌與衍生檢視之間執行分散式事務，要麼需要某種機制，等待事件反映到檢視中。這兩種方法通常都不切實際，因此檢視一般還是非同步更新。

另一方面，從事件日誌衍生當前狀態也簡化了併發控制的某些方面。之所以經常需要多物件事務（參閱 “單物件與多物件操作”），是因為一次使用者操作往往要修改多個不同位置的資料。採用事件溯源後，可以把一個事件設計成對使用者操作的自包含描述。這樣，使用者操作只需在一個位置完成一次寫入——把事件追加到日誌——很容易保證其原子性。

如果事件日誌和應用狀態採用相同的分片方式（例如，處理分片 3 中某位客戶的事件時，只需更新應用狀態的分片 3），那麼簡單的單執行緒日誌消費者無須對寫入做任何併發控制——從設計上看，它一次只會處理一個事件（另見 “實際序列執行”）。日誌在每個分片中定義了事件的序列順序，從而消除了併發帶來的不確定性27。如果一個事件涉及多個狀態分片，就需要多做一些工作，我們將在 第 13 章 中討論。

許多並未採用事件溯源模型的系統，同樣依靠不變性來控制併發：各種資料庫在內部使用不可變資料結構或多版本資料來支援時間點快照（參閱 “索引與快照隔離”）。Git、Mercurial、Fossil 等版本控制系統也依靠不可變資料儲存檔案的版本歷史。

不變性的侷限

永久儲存所有變化的不可變歷史，究竟在多大程度上可行？答案取決於資料集的變動量。有些工作負載以新增資料為主，很少更新或刪除，很容易做成不可變的。另一些工作負載則在相對較小的資料集上頻繁更新和刪除；這時，不可變歷史可能膨脹到難以承受，碎片化也可能成為問題，而壓實和垃圾回收的效能會直接影響系統能否穩健執行45、46。

除了效能原因，有時還必須出於行政或法律原因刪除資料，哪怕這有悖於不變性。例如，歐盟《通用資料保護條例》（GDPR）等隱私法規要求應使用者請求刪除其個人資訊和錯誤資訊；意外洩露敏感資訊後，也可能需要控制影響範圍。

在這些情況下，只在日誌末尾追加一個事件，表示先前的資料應視為已刪除，並不足夠——你真正想做的是改寫歷史，假裝那些資料從未寫入。Datomic 把這種功能稱為 切除（excision）47，Fossil 版本控制系統中也有一個類似概念，稱為 排斥（shunning）48。

真正刪除資料出乎意料地困難49，因為副本可能存在於許多地方。例如，儲存引擎、檔案系統和 SSD 往往把資料寫到新位置，而不是在原地覆蓋41；備份通常還會被刻意設計成不可變，以防意外刪除或損壞。

一種允許刪除不可變資料的方法是 密碼學粉碎（crypto-shredding）50：把將來可能需要刪除的資料加密儲存；需要清除時，忘掉加密金鑰。加密後的資料仍然存在，但已經無人能夠使用。從某種意義上說，這只是轉移了問題：實際資料現在不可變了，儲存金鑰的地方卻是可變的。

此外，還必須事先決定哪些資料共用一把金鑰，何時要改用不同金鑰。這項決定非常重要，因為以後只能選擇粉碎某把金鑰加密的全部資料，或者一項也不粉碎，不能只刪除其中一部分。如果為每個資料項分別儲存一把金鑰，金鑰儲存會變得與主資料儲存一樣龐大，難以管理。可穿刺加密（puncturable encryption）等更複雜的方案51可以選擇性撤銷一把金鑰的部分解密能力，但尚未得到廣泛應用。

總體而言，刪除更像是“讓資料更難取回”，而不是真正“讓資料無法取回”。儘管如此，有時仍必須嘗試，我們將在 “立法與自律” 中看到這一點。

流處理

到目前為止，本章已經討論了流從何而來（使用者活動事件、感測器和資料庫寫入），以及如何傳輸流（直接傳遞訊息、透過訊息代理傳遞，以及使用事件日誌）。

接下來要討論的是，拿到一股流之後能用它做什麼——也就是如何處理它。大體上有三種選擇：

  1. 取出事件中的資料，寫入資料庫、快取、搜尋索引或類似的儲存系統，再供其他客戶端查詢。如 圖 12-5 所示，這是一種讓資料庫與系統其他部分的變化保持同步的好辦法，尤其是在流消費者是唯一寫入資料庫的客戶端時。寫入儲存系統，相當於以流式方式完成 “批處理用例” 中討論的工作。

  2. 以某種方式把事件推送給使用者，例如傳送告警郵件或推送通知，或者把事件流式傳送到實時儀表板上加以視覺化。這種情況下，人是流的最終消費者。

  3. 處理一股或多股輸入流，產生一股或多股輸出流。一股流可能先後經過由多個處理階段組成的流水線，最終才到達某個輸出（即選項 1 或 2）。

本章餘下部分將討論第三種選擇：處理流併產生其他衍生流。執行這種流處理的程式碼稱為 運算元（operator）或 作業（job）。它與 第 11 章 中討論的 Unix 程序和 MapReduce 作業關係密切，資料流模式也很相似：流處理器以只讀方式消費輸入流，再以僅追加方式把輸出寫到另一個位置。

流處理器中的分片和並行化模式，也與 第 11 章 介紹的 MapReduce 和資料流引擎非常相似，因此這裡不再贅述。轉換、過濾記錄等基本對映操作的工作方式也相同。

流與批處理作業有一個關鍵區別：流永遠不會結束。這個區別會帶來許多後果。正如本章開頭所說，對無界資料集進行排序沒有意義，因此不能使用排序合併連線（參閱 “JOIN 與 GROUP BY”）。容錯機制也必須改變：一個只運行了幾分鐘的批處理作業發生任務故障時，大可從頭重啟該任務；但一個流作業已經運行了幾年，崩潰後再從頭開始，通常不可行。

流處理的應用

長期以來，流處理一直用於監控：組織希望在特定事情發生時收到警報。例如：

  • 欺詐檢測系統需要判斷信用卡的使用模式是否出現意外變化，並在信用卡可能被盜時將其凍結。

  • 交易系統需要觀察金融市場的價格變化，並按照指定規則執行交易。

  • 製造系統需要監控工廠內機器的狀態，一旦發生故障便迅速查明問題。

  • 軍事與情報系統需要追蹤潛在侵略者的活動，發現攻擊跡象時發出警報。

這類應用需要相當複雜的模式匹配與關聯分析。不過，流處理也逐漸出現了其他用途。本節將簡要比較其中幾種應用。

複合事件處理

複合事件處理（complex event processing，CEP）是一種在 20 世紀 90 年代發展起來的事件流分析方法，特別適合需要搜尋特定事件模式的應用52。正如正則表示式可以在字串中搜索特定的字元模式，CEP 允許你指定規則，在流中搜索特定的事件模式。

CEP 系統通常使用 SQL 等高階宣告式查詢語言或圖形使用者介面，描述應該檢測哪些事件模式。這些查詢會提交給處理引擎；引擎消費輸入流，並在內部維護一個狀態機來執行所需的匹配。一旦找到匹配，引擎便發出一個 複合事件（名稱由此而來），其中包含檢測到的事件模式詳情53。

這類系統中，查詢與資料的關係恰好和普通資料庫相反。資料庫通常持久儲存資料，把查詢視為臨時物件：查詢到來時，資料庫搜尋與之匹配的資料，查詢完成後便將其忘掉。CEP 引擎卻反過來長期儲存查詢；每個事件到來時，引擎都會檢查迄今所見的事件是否形成了與某個常駐查詢相匹配的模式54。

CEP 的實現包括 Esper、Apama 和 TIBCO StreamBase。Flink、Spark Streaming 等分散式流處理器也支援使用 SQL 對流執行宣告式查詢。

流分析

流處理的另一個用途是對流進行 分析。CEP 與流分析的邊界並不清晰，但一般來說，分析不太關心尋找特定的事件序列，而更關注大量事件上的聚合與統計指標，例如：

  • 測量某類事件的速率（每個時間間隔發生多少次）；

  • 計算某段時間內一個值的滾動平均數；

  • 將當前統計值與先前時間段對比（例如檢測趨勢，或在某項指標與上週同一時間相比異常偏高或偏低時發出警報）。

這類統計值通常在固定時間區間內計算。例如，你可能想知道過去 5 分鐘內某項服務平均每秒收到多少次查詢，以及這段時間內響應時間的第 99 百分位點。在幾分鐘內取平均，可以抹平相鄰秒之間無關緊要的波動，同時仍能及時反映流量模式的變化。用於聚合的時間區間稱為 視窗，我們將在 “時間推理” 中詳細討論。

流分析系統有時會使用機率演算法，例如用布隆過濾器（我們在 “布隆過濾器” 中見過）判斷集合成員關係，用 HyperLogLog55 估計基數，以及用各種演算法估計百分位點（參閱 “計算百分位點”）。機率演算法給出近似結果，但與精確演算法相比，流處理器所需記憶體少得多。近似演算法的這種用途有時讓人誤以為流處理系統總是有損而不精確，其實不然：流處理本身並沒有任何近似性，使用機率演算法只是一項最佳化56。

許多開源分散式流處理框架都以分析為設計目標，例如 Apache Storm、Spark Streaming、Flink、Samza、Apache Beam 和 Kafka Streams57。託管服務則包括 Google Cloud Dataflow 和 Azure Stream Analytics。

維護物化檢視

我們已經看到，資料庫的變更流可以用來維護快取、搜尋索引和資料倉庫等衍生資料系統，使它們與源資料庫保持同步。這些都是維護物化檢視的例子：從某個資料集衍生出另一種檢視，以便高效查詢，並在底層資料變化時更新檢視37。

同樣，在事件溯源中，應用狀態透過應用事件日誌來維護；這裡的應用狀態也是一種物化檢視。與流分析不同，只考慮某個時間視窗內的事件通常不夠：除了日誌壓實可能丟棄的過時事件，構建物化檢視可能需要任意時間段內的 所有 事件。實際上，你需要一個一直延伸到時間開端的視窗。

原則上，任何流處理器都可以用於維護物化檢視。不過，有些面向分析的框架假定自己主要處理持續時間有限的視窗，而永久維護事件與這種假設背道而馳。Kafka Streams 和 Confluent 的 ksqlDB 建立在 Kafka 的日誌壓實支援之上，能夠支援這類用途58。

增量檢視維護

資料庫似乎很適合維護物化檢視——畢竟，它們本來就是用來儲存資料集完整副本的，而且許多資料庫也支援物化檢視。我們在 “物化檢視與多維資料集” 中看到，資料倉庫常見的分析查詢可以物化成 OLAP 多維資料集。

遺憾的是，資料庫通常透過批處理作業，或 PostgreSQL 的 REFRESH MATERIALIZED VIEW 之類的按需請求來重新整理物化視圖表。檢視會定期重新計算，而不是在源資料更新時隨之更新。這種方式有兩個重大缺點，因而不適合透過流處理維護檢視：

  1. 效率低下：每次更新檢視都要重新處理所有資料，儘管絕大多數資料很可能沒有變化。

  2. 資料不夠新鮮：只有等到下一次計劃更新重新執行查詢，源資料的變化才會反映到物化檢視中。

如果資料很容易分割槽，而且計算天然適合增量執行，也可以編寫資料庫觸發器來高效更新物化檢視。例如，如果物化檢視維護每日銷售總收入，那麼每發生一筆新銷售，只需更新相應日期的那一行。在少數場景中可以定製這樣的解決方案，但許多 SQL 查詢很難高效地轉換為增量計算，甚至根本無法轉換。

增量檢視維護（incremental view maintenance，IVM）是解決上述問題的一種更通用方法。IVM 技術把 SQL 等關係語言轉換成能夠執行增量計算的運算元。IVM 演算法不再處理整個資料集，而只重新計算和更新發生變化的資料38、59、60。這樣，檢視計算的效率大幅提升，更新也可以更頻繁地執行，顯著改善資料新鮮度。

Materialize61、RisingWave、ClickHouse 和 Feldera 等資料庫都採用 IVM 技術，提供高效的增量物化檢視。這些資料庫攝取事件流，實時提供物化檢視。最近的事件快取在記憶體中，並定期用於更新磁碟上的物化檢視。讀取時則把最近事件與已經物化的資料合併起來，提供一個統一的實時檢視。由於讀取通常用 SQL 表達，而物化檢視往往以 OLAP 風格的格式儲存，這些系統也支援 第 11 章 所討論的大規模資料倉庫式查詢。

在流上搜索

CEP 可以搜尋由多個事件構成的模式；除此以外，有時還需要按照全文搜尋查詢等複雜條件來搜尋單個事件。

例如，媒體監測服務可以訂閱媒體機構釋出的新聞文章和節目源，搜尋任何提及目標公司、產品或話題的新聞。為此，需要預先制定搜尋查詢，再不斷讓新聞條目流與該查詢進行匹配。一些網站也有類似功能：例如，房地產網站的使用者可以要求網站在市場上出現符合其搜尋條件的新房源時通知他們。Elasticsearch 的 percolator 功能62 就是實現這類流式搜尋的一種選擇。

傳統搜尋引擎先為文件建立索引，再在索引上執行查詢。搜尋資料流卻把這個過程顛倒過來：查詢被儲存下來，文件像 CEP 中的事件一樣逐一流過這些查詢。最簡單的做法是讓每份文件測試每條查詢，但查詢數量很大時會變慢。為了最佳化這個過程，也可以像索引文件那樣索引查詢，從而縮小可能匹配的查詢集合63。

事件驅動架構與 RPC

在 “事件驅動的架構” 中，我們討論過用訊息傳遞系統代替 RPC，也就是把它用作服務之間的通訊機制，Actor 模型便是一例。這些系統同樣以訊息和事件為基礎，但我們通常不會把它們視為流處理器：

  • Actor 框架主要用於管理相互通訊的模組如何併發和分散式執行，而流處理主要是一種資料管理技術。

  • Actor 之間的通訊往往是短暫的一對一通訊，而事件日誌持久存在，並有多個訂閱者。

  • Actor 可以採用任意方式通訊，包括迴圈的請求/響應模式；流處理器通常組成無環流水線，每股流都是某個特定作業的輸出，並從一組明確定義的輸入流衍生而來。

不過，類 RPC 系統與流處理之間也有一些交叉。例如，Apache Storm 有一項稱為 分散式 RPC 的功能，可以把使用者查詢分派給一組同時處理事件流的節點。這些查詢會與輸入流中的事件交錯處理，結果再彙總並返回給使用者（另見 “多分片資料處理”）。

Actor 框架也可以用來處理流。不過，許多這類框架無法保證發生崩潰時訊息仍能送達；除非另外實現重試邏輯，否則處理過程不具備容錯能力。

時間推理

流處理器經常要應對時間問題，尤其是用於分析時，往往會使用“過去五分鐘的平均值”這樣的時間視窗。“過去五分鐘”聽起來似乎清楚明確，實際上卻出人意料地棘手。

批處理任務會迅速處理大量歷史事件。如果需要按時間劃分結果，批處理就必須檢視每個事件中嵌入的時間戳。檢視執行批處理的機器系統時鐘毫無意義，因為任務的執行時間與事件的實際發生時間沒有關係。

一個批處理任務可能幾分鐘內就讀完一整年的歷史事件；大多數情況下，我們關心的是這一年的歷史時間線，而不是幾分鐘的處理時間。此外，使用事件中的時間戳還可以讓處理具有確定性：針對同一輸入重新運行同一個過程，會得到相同的結果。

另一方面，許多流處理框架使用處理機器的本地系統時鐘（即 處理時間）來劃分視窗64。這種方法簡單明了；如果事件建立與事件處理之間的延遲短到可以忽略，也很合理。但只要處理延遲比較顯著——也就是事件實際發生後過了一段明顯可感的時間才得到處理——這種方法就會失效。

事件時間與處理時間

很多原因都會導致處理延遲：排隊、網路故障、效能問題使訊息代理或處理器發生爭用、流消費者重啟，以及故障恢復或修復程式碼缺陷後重新處理過去的事件。

訊息延遲還會使訊息以不可預測的順序到達。例如，假設使用者先發出一個 Web 請求，由 Web 伺服器 A 處理；隨後發出第二個請求，由伺服器 B 處理。A 和 B 各自發出事件，描述自己處理的請求，但 B 的事件先於 A 的事件到達訊息代理。於是，流處理器先看到 B 的事件，再看到 A 的事件，而它們實際發生的順序恰好相反。

不妨拿《星球大戰》系列電影來類比：第四部於 1977 年上映，第五部於 1980 年上映，第六部於 1983 年上映；隨後依次是 1999、2002 和 2005 年上映的第一、二、三部，以及 2015、2017 和 2019 年上映的第七、八、九部65。如果按上映順序觀看，你處理這些電影的順序就與故事的敘事順序不同（集數好比事件時間戳，觀看日期則是處理時間）。人類能夠應對這種不連續性，但流處理演算法必須專門設計，才能處理這類時間與順序問題。

混淆事件時間與處理時間會產生錯誤資料。例如，假設有一個流處理器用來測量請求速率（統計每秒請求數）。重新部署流處理器時，它可能停機一分鐘，恢復執行後再處理積壓的事件。如果按處理時間計算速率，處理積壓期間看起來會突然出現異常的請求尖峰，而真實的請求速率其實一直很穩定（圖 12-8）。

圖 12-8. 按處理時間劃分視窗，會因處理速率的變化而產生人為假象。

處理滯留事件

按事件時間定義視窗時，有一個棘手的問題：你永遠無法確定某個視窗的所有事件是否已經到齊，還是仍有一些事件尚未到達。

例如，假設把事件分成一分鐘的視窗，以統計每分鐘的請求數。你已經統計了一批時間戳落在本小時第 37 分鐘的事件；隨著時間推移，新到事件現在大多落在第 38 和第 39 分鐘。究竟什麼時候才能宣佈第 37 分鐘的視窗已經結束，並輸出計數器的值？

如果一段時間內沒有再看到屬於某個視窗的新事件，可以讓它超時並宣佈視窗就緒。然而，某些事件可能仍快取在另一臺機器上，因網路中斷而延遲。你必須能夠處理這些在視窗宣佈完成後才到達的 滯留事件（straggler event）。大體上有兩種選擇1：

  1. 忽略滯留事件，因為正常情況下，它們可能只佔所有事件的很小一部分。可以把丟棄事件的數量作為指標跟蹤；如果開始丟棄大量資料，就發出警報。

  2. 釋出 更正：為視窗釋出一個包含滯留事件的更新值。可能還需要撤回先前的輸出。

有時可以用一條特殊訊息表示：“從現在起，不會再有時間戳早於 t 的訊息。”消費者可以利用它觸發視窗66。然而，如果多臺機器上的多個生產者都在生成事件，各自擁有不同的最小時間戳閾值，消費者就必須分別跟蹤每個生產者。這時，增加或移除生產者會更加棘手。

你用的是誰的時鐘？

如果事件會在系統中的多個位置緩衝，為事件賦予時間戳就更加困難。例如，考慮一個向伺服器上報使用指標的移動應用。使用者可能在裝置離線時使用該應用；此時應用會把事件快取在裝置本地，等下次連上網際網路時再發送給伺服器，而這可能已是幾小時甚至幾天之後。對於流的消費者來說，這些事件看起來就像延遲極久的滯留事件。

這種情況下，事件時間戳其實應該是使用者交互發生的時間，以移動裝置的本地時鐘為準。然而，使用者控制的裝置時鐘往往不可信，因為它可能被無意或有意地設成錯誤時間（參閱 “時鐘同步和準確性”）。伺服器收到事件的時間以伺服器時鐘為準；由於伺服器在你的控制之下，這個時間更可能準確，卻不能很好地描述使用者互動。

為了校正不準確的裝置時鐘，一種方法是記錄三個時間戳67：

  • 根據裝置時鐘，事件發生的時間；

  • 根據裝置時鐘，事件發往伺服器的時間；

  • 根據伺服器時鐘，伺服器收到事件的時間。

用第三個時間戳減去第二個，可以估算裝置時鐘與伺服器時鐘之間的偏移量（假設相對於所需的時間戳精度，網路延遲可以忽略）。再把這一偏移量應用到事件時間戳上，就能估算事件真正發生的時間——這裡還要假設，從事件發生到事件發往伺服器期間，裝置時鐘的偏移量沒有變化。

這個問題並非流處理獨有，批處理在時間推理上面臨一模一樣的問題。只是在流式環境中，我們更能意識到時間正在流逝，所以問題也更顯眼。

視窗的型別

明確了如何確定事件時間戳之後，下一步就是決定如何定義時間視窗。視窗可用於聚合，例如統計事件數量，或計算視窗內各個值的平均數。以下幾類視窗比較常見64、68：

滾動視窗（tumbling window）

滾動視窗的長度固定，每個事件恰好屬於一個視窗。例如，使用一分鐘的滾動視窗時，時間戳介於 10:03:00 和 10:03:59 的所有事件歸入一個視窗，介於 10:04:00 和 10:04:59 的事件歸入下一個視窗，依此類推。實現一分鐘滾動視窗時，可以把每個事件的時間戳向下取整到最近的整分鐘，以確定它屬於哪個視窗。

跳躍視窗（hopping window）

跳躍視窗的長度也固定，但允許視窗重疊，以起到一定的平滑作用。例如，一個長度為五分鐘、跳躍步長為一分鐘的視窗，先包含 10:03:00 到 10:07:59 之間的事件；下一個視窗包含 10:04:00 到 10:08:59 之間的事件，依此類推。可以先計算一分鐘的滾動視窗，再聚合相鄰的多個視窗，從而實現這個跳躍視窗。

滑動視窗（sliding window）

滑動視窗包含彼此間隔不超過某段時間的所有事件。例如，五分鐘的滑動視窗會同時覆蓋發生在 10:03:39 和 10:08:12 的事件，因為兩者相差不到五分鐘。注意，五分鐘的滾動視窗或跳躍視窗使用固定邊界，不一定會把這兩個事件放在同一個視窗中。實現滑動視窗時，可以維護一個按時間排序的事件緩衝區，並在舊事件過期、離開視窗時將其移除。

會話視窗（session window）

會話視窗與其他視窗不同，沒有固定的持續時間。它把同一使用者在時間上彼此接近的事件歸為一組，並在使用者有一段時間沒有活動後結束視窗（例如，30 分鐘內沒有任何事件）。網站分析經常需要劃分會話。

視窗操作通常需要維護臨時狀態。有些情況下，無論視窗多大、發生多少事件，狀態的大小都是固定的：例如，計數操作不論視窗大小和事件數量如何，都只需要一個計數器。另一方面，滑動視窗以及下一節討論的流連線，都必須緩衝事件，直到視窗結束。因此，視窗很大或流吞吐量很高時，流處理器可能需要儲存大量臨時狀態。無論這些狀態儲存在記憶體還是磁碟上，都必須確保執行流處理任務的機器有足夠容量來容納它們。

流連線

在 “JOIN 與 GROUP BY” 中，我們討論過批處理作業如何按鍵連線資料集，以及這種連線為何是資料流水線的重要組成部分。流處理把資料流水線推廣到對無界資料集的增量處理，因此也同樣需要對流執行連線。

不過，流中隨時可能出現新事件，使流連線比批處理作業中的連線更具挑戰。為了看清這個問題，我們把連線分成三類：流—流連線、流—表連線 和 表—表連線。下面各用一個例子來說明。

流流連線（視窗連線）

假設網站提供搜尋功能，而你想發現最近的 URL 搜尋趨勢。每當有人輸入搜尋查詢，就記錄一個包含查詢及返回結果的事件；每當有人點選某項搜尋結果，又記錄一個點選事件。為了計算搜尋結果中每個 URL 的點選率，必須把搜尋行為與點選行為的事件結合起來；它們可以透過相同的會話 ID 關聯。廣告系統也需要類似的分析69。

如果使用者放棄搜尋，點選也許永遠不會發生；即使發生，搜尋與點選之間的間隔也可能相差懸殊：通常只有幾秒，但也可能長達數天或數週——例如使用者搜尋之後忘記了這個瀏覽器標籤頁，過了很久才回來點選某個結果。網路延遲不一，甚至可能讓點選事件先於搜尋事件到達。你可以為連線選擇適當的視窗，例如只連線相隔不超過一小時的搜尋與點選。

請注意，把搜尋詳情嵌入點選事件並不等同於連線兩類事件：這樣只能瞭解使用者點選搜尋結果的情況，卻無法瞭解使用者沒有點選任何結果的搜尋。衡量搜尋質量需要準確的點選率，因此搜尋事件和點選事件缺一不可。

為了實現這類連線，流處理器需要維護 狀態，例如按會話 ID 索引過去一小時內的所有事件。每當搜尋事件或點選事件到來，就把它加入相應索引，同時檢查另一個索引，看看相同會話 ID 的另一事件是否已經到達。找到匹配時，發出一個事件，說明哪項搜尋結果被點選；如果搜尋事件過期時仍未看到匹配的點選事件，則發出一個事件，說明哪些搜尋結果沒有被點選。

流表連線（流擴充）

在 “JOIN 與 GROUP BY”（圖 11-2）中，我們見過批處理作業連線兩個資料集的例子：一組使用者活動事件和一個使用者檔案資料庫。很自然地，可以把使用者活動事件視為一股流，在流處理器中持續執行同樣的連線：輸入是包含使用者 ID 的活動事件流，輸出則是活動事件流，其中的使用者 ID 已經補充了相應的使用者檔案資訊。這個過程有時稱為用資料庫中的資訊 擴充（enrich）活動事件。

執行這項連線時，流處理器要逐一檢視活動事件，在資料庫中查詢事件裡的使用者 ID，再把檔案資訊加入活動事件。資料庫查詢可以透過查詢遠端資料庫來實現；不過，正如 “JOIN 與 GROUP BY” 中所討論的，這類遠端查詢很可能速度緩慢，還可能使資料庫過載58。

另一種做法是把資料庫副本載入流處理器，在本地查詢，免去網路往返。由於資料庫的本地副本可能是記憶體散列表（如果足夠小），也可能是本地磁碟上的索引，因此這項技術稱為 雜湊連線。

它與批處理作業的區別在於：批處理作業把資料庫某個時間點的快照用作輸入；流處理器卻長期執行，而資料庫內容很可能隨時間變化，所以流處理器中的本地副本必須持續更新。變更資料捕獲可以解決這個問題：除了活動事件流，流處理器還可以訂閱使用者檔案資料庫的變更日誌。每當建立或修改檔案時，流處理器就更新本地副本。這樣，我們實際上得到了兩股流之間的連線：活動事件與檔案更新。

流—表連線其實與流—流連線很相似。最大的區別在於，對錶的變更日誌流執行連線時，使用的是一個回溯至“時間開端”的視窗（概念上是無限視窗），其中記錄的新版本會覆蓋舊版本；對另一股輸入流，連線可能根本不維護視窗。

表表連線（維護物化檢視）

考慮 “案例研究：社交網路首頁時間線” 中討論的社交網路時間線。我們說過，使用者檢視主頁時間線時，如果遍歷他所關注的所有人，找出他們最近釋出的帖子再合併，代價實在太高。

我們需要的是時間線快取：為每位使用者準備一個“收件箱”，帖子釋出時便寫入其中，這樣讀取時間線只需查詢一次。物化並維護這個快取，需要處理以下事件：

  • 使用者 u 釋出新帖子時，把帖子加入所有關注 u 的使用者的時間線。

  • 使用者刪除一篇帖子或刪除整個賬戶時，從所有使用者的時間線中移除相應帖子。

  • 使用者 u1 開始關注使用者 u2 時，把 u2 最近的帖子加入 u1 的時間線。

  • 使用者 u1 取消關注使用者 u2 時，從 u1 的時間線中移除 u2 的帖子。

要在流處理器中維護這項快取，需要一股帖子事件流（釋出和刪除），以及一股關注關係事件流（關注和取消關注）。流處理器還要維護一個數據庫，記錄每位使用者的關注者集合，以便新帖子到來時知道應該更新哪些時間線。

也可以換個角度來看：這項流處理維護著一個連線兩張表（帖子和關注關係）的查詢物化檢視，大致如下：

SELECT follows.follower_id AS timeline_id,
  array_agg(posts.* ORDER BY posts.timestamp DESC)
FROM posts
JOIN follows ON follows.followee_id = posts.sender_id
GROUP BY follows.follower_id

流之間的連線直接對應查詢中的表連線。時間線實際上就是查詢結果的快取，每當底層表發生變化時都會更新。

Note

如果像 圖 12-7 那樣，把流視為表的導數，再把連線視為兩個表的乘積 u·v，就會出現一個有趣的結果：物化連線的變更流遵循乘積法則 (u·v)′ = u′v + uv′。換句話說，帖子的任何變化都要與當前關注者連線，而關注者的任何變化都要與當前帖子連線37。

連線的時間依賴性

這裡介紹的三類連線（流—流、流—表和表—表）有許多共同點：流處理器都要根據連線的一側維護某種狀態（搜尋和點選事件、使用者檔案或關注者列表），再在連線另一側的訊息到來時查詢該狀態。

維護狀態的事件順序十分重要——先關注再取消關注，與順序相反的結果不同。在 Kafka 這樣的分片事件日誌中，同一個分片（分割槽）內的事件順序能夠保持，但不同流或不同分片之間通常沒有順序保證。

這就帶來一個問題：如果不同流上的事件發生時間相近，應該按什麼順序處理？在流—表連線的例子中，如果使用者更新了檔案，哪些活動事件應該與舊檔案連線（在檔案更新前處理），哪些應該與新檔案連線（在檔案更新後處理）？換句話說，如果狀態隨時間變化，而你要與這個狀態連線，究竟應該使用哪個時間點的狀態？

這種時間依賴會出現在許多地方。例如，銷售商品時要為發票採用正確的稅率；稅率取決於國家或州、產品型別和銷售日期，因為稅率會不時變化。把銷售記錄與稅率表連線時，通常需要採用銷售發生時的稅率；如果正在重新處理歷史資料，它可能與當前稅率不同。

如果不同流之間的事件順序不確定，連線也會變得不確定70。這意味著，即使針對同一輸入重新運行同一個作業，也不一定得到相同結果：再次執行時，各輸入流上的事件可能以不同方式交錯。

在資料倉庫中，這個問題稱為 緩慢變化維度（slowly changing dimension，SCD），通常為所連線記錄的每個具體版本賦予唯一識別符號來解決。例如，每次稅率變化時，都給新稅率分配一個新識別符號；發票則包含銷售時所用稅率的識別符號71、72。這樣連線就具有確定性，但也無法再做日誌壓實，因為必須保留表中記錄的所有版本。另一種方法是反正規化資料，把適用稅率直接放入每個銷售事件中。

容錯

本章最後一節來看看流處理器如何容忍故障。我們在 第 11 章 中看到，批處理框架相當容易實現容錯：一項任務失敗後，只須在另一臺機器上重新啟動，並丟棄失敗任務的輸出。之所以能透明地重試，是因為輸入檔案不可變，每項任務都把輸出寫入獨立檔案，而且只有任務成功完成後，輸出才會變得可見。

具體來說，批處理的容錯方式可以確保：即使某些任務實際上失敗過，批處理作業的輸出也與沒有發生過任何問題時相同。看起來每條輸入記錄都只處理了恰好一次——沒有記錄被跳過，也沒有記錄被處理兩遍。重新啟動任務意味著記錄實際上可能處理了多次，但輸出中可見的效果卻像只處理了一次。這個原則稱為 恰好一次語義（exactly-once semantics），不過 等效一次（effectively-once）或許是更貼切的叫法73。

流處理也面臨同樣的容錯問題，卻沒那麼容易解決：不能等任務完成後才讓輸出可見，因為流是無限的，永遠也處理不完。

微批處理與檢查點

一種解決辦法是把流分成小塊，把每一塊當作一個微型批處理。這種方法稱為 微批處理（microbatching），Spark Streaming 就採用了它74。批次通常約為一秒，這是效能權衡的結果：批次越小，排程與協調開銷越大；批次越大，流處理器的結果變得可見之前，延遲就越長。

微批處理還隱式提供了一個大小等於批次大小的滾動視窗——它按處理時間而非事件時間戳劃分。需要更大視窗的作業，必須顯式地把狀態從一個微批次傳遞到下一個微批次。

Apache Flink 採用一種變體：定期生成滾動的狀態檢查點，並將其寫入持久儲存75、76。如果流運算元崩潰，可以從最近的檢查點重新啟動，並丟棄從上一個檢查點到崩潰之間產生的所有輸出。檢查點由訊息流中的屏障觸發，類似於微批次之間的邊界，但不會強制規定視窗大小。

在流處理框架內部，微批處理與檢查點都能提供與批處理相同的恰好一次語義。然而，一旦輸出離開流處理器——例如寫入資料庫、向外部訊息代理傳送訊息或傳送電子郵件——框架便無法丟棄失敗微批次的輸出。這時，重啟失敗任務會使外部副作用發生兩次，僅靠微批處理或檢查點不足以避免這個問題。

再談原子提交

為了在發生故障時營造恰好一次處理的效果，必須確保處理事件產生的所有輸出和副作用，當且僅當 處理成功時才生效。這些影響包括：發給下游運算元或外部訊息傳遞系統的訊息（包括電子郵件或推送通知）、資料庫寫入、運算元狀態的變化，以及對輸入訊息的確認應答（包括推進基於日誌的訊息代理中的消費者偏移量）。

這些事情要麼全部以原子方式發生，要麼一件也不發生，不能彼此失去同步。這種做法聽起來似曾相識，是因為我們曾在分散式事務和兩階段提交的語境下討論過它（參閱 “恰好一次訊息處理”）。

我們在 第 10 章 中討論了 XA 等傳統分散式事務實現的問題。不過，在限制更嚴格的環境中，可以高效實現這樣的原子提交機制。Google Cloud Dataflow66、75、VoltDB77 和 Apache Kafka78、79 都採用了這種方法。與 XA 不同，這些實現不會試圖跨異構技術提供事務，而是由流處理框架同時管理狀態變化與訊息傳遞，把事務留在框架內部。還可以在單個事務中處理多條輸入訊息，從而分攤事務協議的開銷。

冪等性

我們的目標是丟棄所有失敗任務的部分輸出，以便能安全地重試，而不會生效兩次。分散式事務是實現這個目標的一種方法；另一種方法是依靠 冪等性，正如 “持久化執行與工作流” 中所見80。

冪等操作可以執行多次，效果與只執行一次相同。例如，刪除鍵值儲存中的一個鍵是冪等的（再次刪除不會產生進一步影響）；遞增計數器卻不是冪等的（再執行一次遞增，值就增加了兩次）。

即使操作本身並非天然冪等，通常也可以用一點額外元資料讓它變得冪等。例如，消費 Kafka 訊息時，每條訊息都有一個持久、單調遞增的偏移量。向外部資料庫寫入值時，可以把觸發最近一次寫入的訊息偏移量與值一併寫入。這樣便能判斷某項更新是否已經應用，避免再次執行同一更新。

Storm 的 Trident 也用類似思路處理狀態。依靠冪等性包含幾個假設：重啟失敗任務時，必須以相同順序重播相同訊息（基於日誌的訊息代理可以做到）；處理過程必須具有確定性；而且不能有其他節點併發更新同一個值81、82。

從一個處理節點故障切換到另一個節點時，可能還需要採用柵欄機制（參閱 “分散式鎖和租約”），以防某個被認為已經死亡、實際仍然存活的節點造成干擾。儘管有這麼多限制，冪等操作仍是實現恰好一次語義的有效方式，而且開銷很小。

失敗後重建狀態

任何需要狀態的流處理過程——例如計數器、平均值、直方圖等視窗聚合，以及連線所用的表和索引——都必須確保發生故障後能夠恢復狀態。

一種選擇是把狀態儲存在遠端資料儲存中並進行復制，不過為每條訊息查詢一次遠端資料庫可能很慢。另一種選擇是把狀態儲存在流處理器本地，再定期複製。這樣，流處理器從故障中恢復時，新任務可以讀取狀態副本，繼續處理而不丟失資料。

例如，Flink 定期捕獲運算元狀態快照，將其寫入分散式檔案系統等持久儲存75、76；Kafka Streams 則把狀態變化傳送到一個啟用了日誌壓實的專用 Kafka 主題，以類似變更資料捕獲的方式複製狀態83。VoltDB 會在多個節點上冗餘處理每條輸入訊息，以此複製狀態（參閱 “實際序列執行”）。

有些情況下，甚至不必複製狀態，因為可以根據輸入流重建。例如，如果狀態只是較短視窗上的聚合，重播該視窗對應的輸入事件也許足夠快。如果狀態是由變更資料捕獲維護的資料庫本地副本，也可以從經過日誌壓實的變更流重建資料庫。

不過，所有這些權衡都取決於底層基礎設施的效能特徵：有些系統的網路延遲可能低於磁碟訪問延遲，網路頻寬也可能與磁碟頻寬相當。不存在適用於所有情況的理想權衡；隨著儲存和網路技術演進，本地狀態與遠端狀態各自的優勢也可能發生變化。

本章小結

本章討論了事件流、它們的用途，以及如何處理它們。從某種意義上說，流處理與 第 11 章 討論的批處理非常相似，只不過流處理持續作用於無界（永不終止）的流，而不是大小固定的輸入84。從這個角度看，訊息代理和事件日誌就是檔案系統的流式對應物。

我們花了一些篇幅比較兩類訊息代理：

AMQP/JMS 風格的訊息代理

代理把每條訊息分別分配給消費者；消費者成功處理一條訊息後，就對它進行確認應答。訊息得到確認後，代理會將其刪除。這種方法適合用作非同步 RPC（另見 “事件驅動的架構”），例如用於任務佇列：訊息處理的確切順序並不重要，處理完成後也無須返回重讀舊訊息。

基於日誌的訊息代理

代理把一個分片中的所有訊息分配給同一個消費者節點，並始終按相同順序傳遞訊息。系統透過分片實現並行；消費者則為已處理的最後一條訊息偏移量建立檢查點，以此跟蹤進度。代理把訊息保留在磁碟上，因此必要時可以跳回去重新讀取舊訊息。

基於日誌的方法與資料庫的複製日誌（參閱 第 6 章）和日誌結構儲存引擎（參閱 第 4 章）有相似之處。正如 第 10 章 所述，它也是共識的一種形式。我們看到，這種方法尤其適合那些消費輸入流，並生成衍生狀態或衍生輸出流的流處理系統。

至於流從何而來，我們討論了幾種可能：使用者活動事件、定期提供讀數的感測器，以及資料來源（例如金融市場資料），都天然適合表示成流。把資料庫寫入視為流同樣很有用：可以透過變更資料捕獲隱式取得變更日誌——即資料庫所有變化的歷史——也可以透過事件溯源顯式取得。日誌壓實讓流可以保留資料庫內容的完整副本。

把資料庫表示成流，為系統整合帶來了強大能力。消費變更日誌並把變化應用到衍生系統，就能讓搜尋索引、快取、分析系統等衍生資料系統持續保持最新。甚至可以從頭開始消費變更日誌，一直追到當前時刻，從而基於現有資料構建全新的檢視。

以流的形式維護狀態、重播訊息的能力，也是各種流處理框架實現流連線和容錯技術的基礎。我們討論了流處理的幾種用途，包括搜尋事件模式（複合事件處理）、計算視窗聚合（流分析），以及讓衍生資料系統保持最新（物化檢視）。

隨後，我們討論了流處理器進行時間推理時的困難，包括處理時間與事件時間戳的區別，以及如何處理那些在視窗本以為已經完成後才姍姍來遲的滯留事件。

我們區分了流處理過程中可能出現的三類連線：

流—流連線

兩股輸入流都由活動事件組成，連線運算元會在某個時間視窗內尋找彼此相關的事件。例如，它可以匹配同一使用者在 30 分鐘內執行的兩個操作。如果想尋找同一股流中的相關事件，連線的兩側實際上也可以是同一股流，這稱為 自連線（self-join）。

流—表連線

一股輸入流由活動事件組成，另一股則是資料庫變更日誌。變更日誌用來讓資料庫的本地副本保持最新。對於每個活動事件，連線運算元查詢資料庫，並輸出經過擴充的活動事件。

表—表連線

兩股輸入流都是資料庫變更日誌。這種情況下，任意一側的每次變化都與另一側的最新狀態連線。結果是兩張表連線所得物化檢視的變更流。

最後，我們討論了流處理器實現容錯和恰好一次語義的技術。與批處理一樣，必須丟棄失敗任務產生的部分輸出。但流處理過程長期執行並持續產生輸出，無法簡單地丟棄所有輸出。因此，需要微批處理、檢查點、事務或冪等寫入等機制，在更細的粒度上進行恢復。

腳註

參考文獻

  1. Tyler Akidau, Robert Bradshaw, Craig Chambers, Slava Chernyak, Rafael J. Fernández-Moctezuma, Reuven Lax, Sam McVeety, Daniel Mills, Frances Perry, Eric Schmidt, and Sam Whittle. The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing. Proceedings of the VLDB Endowment, volume 8, issue 12, pages 1792–1803, August 2015. doi:10.14778/2824032.2824076 ↩︎ ↩︎

  2. Harold Abelson, Gerald Jay Sussman, and Julie Sussman. Structure and Interpretation of Computer Programs, 2nd edition. MIT Press, 1996. ISBN: 978-0-262-51087-5, archived at archive.org/details/sicp_20211010 ↩︎

  3. Patrick Th. Eugster, Pascal A. Felber, Rachid Guerraoui, and Anne-Marie Kermarrec. The Many Faces of Publish/Subscribe. ACM Computing Surveys, volume 35, issue 2, pages 114–131, June 2003. doi:10.1145/857076.857078 ↩︎

  4. Don Carney, Uğur Çetintemel, Mitch Cherniack, Christian Convey, Sangdon Lee, Greg Seidman, Michael Stonebraker, Nesime Tatbul, and Stan Zdonik. Monitoring Streams – A New Class of Data Management Applications. At 28th International Conference on Very Large Data Bases (VLDB), August 2002. doi:10.1016/B978-155860869-6/50027-5 ↩︎

  5. Matthew Sackman. Pushing Back. wellquite.org, May 2016. Archived at perma.cc/3KCZ-RUFY ↩︎ ↩︎

  6. Thomas Figg (tef). how (not) to write a pipeline. cohost.org, June 2023. Archived at perma.cc/A3V8-NYCM ↩︎

  7. Vicent Martí. Brubeck, a statsd-Compatible Metrics Aggregator. github.blog, June 2015. Archived at perma.cc/TP3Q-DJYM ↩︎

  8. Seth Lowenberger. MoldUDP64 Protocol Specification V 1.00. nasdaqtrader.com, July 2009. Archived at https://perma.cc/7CRQ-QBD7 ↩︎

  9. Ian Malpass. Measure Anything, Measure Everything. codeascraft.com, February 2011. Archived at archive.org ↩︎

  10. Dieter Plaetinck. 25 Graphite, Grafana and statsd Gotchas. grafana.com, March 2016. Archived at perma.cc/3NP3-67U7 ↩︎

  11. Jeff Lindsay. Web Hooks to Revolutionize the Web. progrium.com, May 2007. Archived at perma.cc/BF9U-XNX4 ↩︎

  12. Jim N. Gray. Queues Are Databases. Microsoft Research Technical Report MSR-TR-95-56, December 1995. Archived at arxiv.org ↩︎

  13. Mark Hapner, Rich Burridge, Rahul Sharma, Joseph Fialli, Kate Stout, and Nigel Deakin. JSR-343 Java Message Service (JMS) 2.0 Specification. jms-spec.java.net, March 2013. Archived at perma.cc/E4YG-46TA ↩︎

  14. Sanjay Aiyagari, Matthew Arrott, Mark Atwell, Jason Brome, Alan Conway, Robert Godfrey, Robert Greig, Pieter Hintjens, John O’Hara, Matthias Radestock, Alexis Richardson, Martin Ritchie, Shahrokh Sadjadi, Rafael Schloming, Steven Shaw, Martin Sustrik, Carl Trieloff, Kim van der Riet, and Steve Vinoski. AMQP: Advanced Message Queuing Protocol Specification. Version 0-9-1, November 2008. Archived at perma.cc/6YJJ-GM9X ↩︎

  15. Architectural overview of Pub/Sub. cloud.google.com, 2025. Archived at perma.cc/VWF5-ABP4 ↩︎ ↩︎

  16. Aris Tzoumas. Lessons from scaling PostgreSQL queues to 100k events per second. rudderstack.com, July 2025. Archived at perma.cc/QD8C-VA4Y ↩︎

  17. Robin Moffatt. Kafka Connect Deep Dive – Error Handling and Dead Letter Queues. confluent.io, March 2019. Archived at perma.cc/KQ5A-AB28 ↩︎

  18. Dunith Danushka. Message reprocessing: How to implement the dead letter queue. redpanda.com. Archived at perma.cc/R7UB-WEWF ↩︎

  19. Damien Gasparina, Loic Greffier, and Sebastien Viale. KIP-1034: Dead letter queue in Kafka Streams. cwiki.apache.org, April 2024. Archived at perma.cc/3VXV-QXAN ↩︎

  20. Jay Kreps, Neha Narkhede, and Jun Rao. Kafka: A Distributed Messaging System for Log Processing. At 6th International Workshop on Networking Meets Databases (NetDB), June 2011. Archived at perma.cc/CSW7-TCQ5 ↩︎

  21. Jay Kreps. Benchmarking Apache Kafka: 2 Million Writes Per Second (On Three Cheap Machines). engineering.linkedin.com, April 2014. Archived at archive.org ↩︎

  22. Kartik Paramasivam. How We’re Improving and Advancing Kafka at LinkedIn. engineering.linkedin.com, September 2015. Archived at perma.cc/3S3V-JCYJ ↩︎

  23. Philippe Dobbelaere and Kyumars Sheykh Esmaili. Kafka versus RabbitMQ: A comparative study of two industry reference publish/subscribe implementations. At 11th ACM International Conference on Distributed and Event-based Systems (DEBS), June 2017. doi:10.1145/3093742.3093908 ↩︎

  24. Kate Holterhoff. Why Message Queues Endure: A History. redmonk.com, December 2024. Archived at perma.cc/6DX8-XK4W ↩︎

  25. Andrew Schofield. KIP-932: Queues for Kafka. cwiki.apache.org, May 2023. Archived at perma.cc/LBE4-BEMK ↩︎

  26. Jack Vanlightly. The advantages of queues on logs. jack-vanlightly.com, October 2023. Archived at perma.cc/WJ7V-287K ↩︎

  27. Jay Kreps. The Log: What Every Software Engineer Should Know About Real-Time Data’s Unifying Abstraction. engineering.linkedin.com, December 2013. Archived at perma.cc/2JHR-FR64 ↩︎ ↩︎

  28. Andy Hattemer. Change Data Capture is having a moment. Why? materialize.com, September 2021. Archived at perma.cc/AL37-P53C ↩︎

  29. Prem Santosh Udaya Shankar. Streaming MySQL Tables in Real-Time to Kafka. engineeringblog.yelp.com, August 2016. Archived at perma.cc/5ZR3-2GVV ↩︎

  30. Andreas Andreakis, Ioannis Papapanagiotou. DBLog: A Watermark Based Change-Data-Capture Framework. October 2020. Archived at arxiv.org ↩︎

  31. Jiri Pechanec. Percolator. debezium.io, October 2021. Archived at perma.cc/EQ8E-W6KQ ↩︎

  32. Debezium maintainers. Debezium Connector for Cassandra. debezium.io. Archived at perma.cc/WR6K-EKMD ↩︎

  33. Neha Narkhede. Announcing Kafka Connect: Building Large-Scale Low-Latency Data Pipelines. confluent.io, February 2016. Archived at perma.cc/8WXJ-L6GF ↩︎ ↩︎

  34. Chris Riccomini. Kafka change data capture breaks database encapsulation. cnr.sh, November 2018. Archived at perma.cc/P572-9MKF ↩︎

  35. Gunnar Morling. “Change Data Capture Breaks Encapsulation”. Does it, though? decodable.co, November 2023. Archived at perma.cc/YX2P-WNWR ↩︎

  36. Gunnar Morling. Revisiting the Outbox Pattern. decodable.co, October 2024. Archived at perma.cc/M5ZL-RPS9 ↩︎

  37. Ashish Gupta and Inderpal Singh Mumick. Maintenance of Materialized Views: Problems, Techniques, and Applications. IEEE Data Engineering Bulletin, volume 18, issue 2, pages 3–18, June 1995. Archived at archive.org ↩︎ ↩︎ ↩︎

  38. Mihai Budiu, Tej Chajed, Frank McSherry, Leonid Ryzhyk, Val Tannen. DBSP: Incremental Computation on Streams and Its Applications to Databases. SIGMOD Record, volume 53, issue 1, pages 87–95, March 2024. doi:10.1145/3665252.3665271 ↩︎ ↩︎

  39. Jim Gray and Andreas Reuter. Transaction Processing: Concepts and Techniques. Morgan Kaufmann, 1992. ISBN: 9781558601901 ↩︎

  40. Martin Kleppmann. Accounting for Computer Scientists. martin.kleppmann.com, March 2011. Archived at perma.cc/9EGX-P38N ↩︎

  41. Pat Helland. Immutability Changes Everything. At 7th Biennial Conference on Innovative Data Systems Research (CIDR), January 2015. ↩︎ ↩︎

  42. Martin Kleppmann. Making Sense of Stream Processing. Report, O’Reilly Media, May 2016. Archived at perma.cc/RAY4-JDVX ↩︎

  43. Kartik Paramasivam. Stream Processing Hard Problems – Part 1: Killing Lambda. engineering.linkedin.com, June 2016. Archived at archive.org ↩︎

  44. Stéphane Derosiaux. CQRS: What? Why? How? sderosiaux.medium.com, September 2019. Archived at perma.cc/FZ3U-HVJ4 ↩︎

  45. Baron Schwartz. Immutability, MVCC, and Garbage Collection. xaprb.com, December 2013. Archived at archive.org ↩︎

  46. Daniel Eloff, Slava Akhmechet, Jay Kreps, et al. Re: Turning the Database Inside-out with Apache Samza. Hacker News discussion, news.ycombinator.com, March 2015. Archived at perma.cc/ML9E-JC83 ↩︎

  47. Datomic Documentation: Excision. Cognitect, Inc., docs.datomic.com. Archived at perma.cc/J5QQ-SH32 ↩︎

  48. Fossil Documentation: Deleting Content from Fossil. fossil-scm.org, 2025. Archived at perma.cc/DS23-GTNG ↩︎

  49. Jay Kreps. The irony of distributed systems is that data loss is really easy but deleting data is surprisingly hard. x.com, March 2015. Archived at perma.cc/7RRZ-V7B7 ↩︎

  50. Brent Robinson. Crypto shredding: How it can solve modern data retention challenges. medium.com, January 2019. Archived at https://perma.cc/4LFK-S6XE ↩︎

  51. Matthew D. Green and Ian Miers. Forward Secure Asynchronous Messaging from Puncturable Encryption. At IEEE Symposium on Security and Privacy, May 2015. doi:10.1109/SP.2015.26 ↩︎

  52. David C. Luckham. What’s the Difference Between ESP and CEP? complexevents.com, June 2019. Archived at perma.cc/E7PZ-FDEF ↩︎

  53. Arvind Arasu, Shivnath Babu, and Jennifer Widom. The CQL Continuous Query Language: Semantic Foundations and Query Execution. The VLDB Journal, volume 15, issue 2, pages 121–142, June 2006. doi:10.1007/s00778-004-0147-z ↩︎

  54. Julian Hyde. Data in Flight: How Streaming SQL Technology Can Help Solve the Web 2.0 Data Crunch. ACM Queue, volume 7, issue 11, December 2009. doi:10.1145/1661785.1667562 ↩︎

  55. Philippe Flajolet, Éric Fusy, Olivier Gandouet, and Frédéric Meunier. HyperLogLog: The Analysis of a Near-Optimal Cardinality Estimation Algorithm. At Conference on Analysis of Algorithms (AofA), June 2007. doi:10.46298/dmtcs.3545 ↩︎

  56. Jay Kreps. Questioning the Lambda Architecture. oreilly.com, July 2014. Archived at perma.cc/2WY5-HC8Y ↩︎

  57. Ian Reppel. An Overview of Apache Streaming Technologies. ianreppel.org, March 2016. Archived at perma.cc/BB3E-QJLW ↩︎

  58. Jay Kreps. Why Local State is a Fundamental Primitive in Stream Processing. oreilly.com, July 2014. Archived at perma.cc/P8HU-R5LA ↩︎ ↩︎

  59. RisingWave Labs. Deep Dive Into the RisingWave Stream Processing Engine - Part 2: Computational Model. risingwave.com, November 2023. Archived at perma.cc/LM74-XDEL ↩︎

  60. Frank McSherry, Derek G. Murray, Rebecca Isaacs, and Michael Isard. Differential dataflow. At 6th Biennial Conference on Innovative Data Systems Research (CIDR), January 2013. ↩︎

  61. Andy Hattemer. Incremental Computation in the Database. materialize.com, March 2020. Archived at perma.cc/AL94-YVRN ↩︎

  62. Shay Banon. Percolator. elastic.co, February 2011. Archived at perma.cc/LS5R-4FQX ↩︎

  63. Alan Woodward and Martin Kleppmann. Real-Time Full-Text Search with Luwak and Samza. martin.kleppmann.com, April 2015. Archived at perma.cc/2U92-Q7R4 ↩︎

  64. Tyler Akidau. The World Beyond Batch: Streaming 102. oreilly.com, January 2016. Archived at perma.cc/4XF9-8M2K ↩︎ ↩︎

  65. Stephan Ewen. Streaming Analytics with Apache Flink. At Kafka Summit, April 2016. Archived at perma.cc/QBQ4-F9MR ↩︎

  66. Tyler Akidau, Alex Balikov, Kaya Bekiroğlu, Slava Chernyak, Josh Haberman, Reuven Lax, Sam McVeety, Daniel Mills, Paul Nordstrom, and Sam Whittle. MillWheel: Fault-Tolerant Stream Processing at Internet Scale. Proceedings of the VLDB Endowment, volume 6, issue 11, pages 1033–1044, August 2013. doi:10.14778/2536222.2536229 ↩︎ ↩︎

  67. Alex Dean. Improving Snowplow’s Understanding of Time. snowplow.io, September 2015. Archived at perma.cc/6CT9-Z3Q2 ↩︎

  68. Azure Stream Analytics: Windowing functions. Microsoft Azure Reference, learn.microsoft.com, July 2025. Archived at archive.org ↩︎

  69. Rajagopal Ananthanarayanan, Venkatesh Basker, Sumit Das, Ashish Gupta, Haifeng Jiang, Tianhao Qiu, Alexey Reznichenko, Deomid Ryabkov, Manpreet Singh, and Shivakumar Venkataraman. Photon: Fault-Tolerant and Scalable Joining of Continuous Data Streams. At ACM International Conference on Management of Data (SIGMOD), June 2013. doi:10.1145/2463676.2465272 ↩︎

  70. Ben Kirwin. Doing the Impossible: Exactly-Once Messaging Patterns in Kafka. ben.kirw.in, November 2014. Archived at perma.cc/A5QL-QRX7 ↩︎

  71. Pat Helland. Data on the Outside Versus Data on the Inside. At 2nd Biennial Conference on Innovative Data Systems Research (CIDR), January 2005. ↩︎

  72. Ralph Kimball and Margy Ross. The Data Warehouse Toolkit: The Definitive Guide to Dimensional Modeling, 3rd edition. John Wiley & Sons, 2013. ISBN: 978-1-118-53080-1 ↩︎

  73. Viktor Klang. I’m coining the phrase ’effectively-once’ for message processing with at-least-once + idempotent operations. x.com, October 2016. Archived at perma.cc/7DT9-TDG2 ↩︎

  74. Matei Zaharia, Tathagata Das, Haoyuan Li, Scott Shenker, and Ion Stoica. Discretized Streams: An Efficient and Fault-Tolerant Model for Stream Processing on Large Clusters. At 4th USENIX Conference in Hot Topics in Cloud Computing (HotCloud), June 2012. ↩︎

  75. Kostas Tzoumas, Stephan Ewen, and Robert Metzger. High-Throughput, Low-Latency, and Exactly-Once Stream Processing with Apache Flink. ververica.com, August 2015. Archived at archive.org ↩︎ ↩︎ ↩︎

  76. Paris Carbone, Gyula Fóra, Stephan Ewen, Seif Haridi, and Kostas Tzoumas. Lightweight Asynchronous Snapshots for Distributed Dataflows. arXiv:1506.08603

    cs.DCcs.DC

    , June 2015. ↩︎ ↩︎

  77. Ryan Betts and John Hugg. Fast Data: Smart and at Scale. Report, O’Reilly Media, October 2015. Archived at perma.cc/VQ6S-XQQY ↩︎

  78. Neha Narkhede and Guozhang Wang. Exactly-Once Semantics Are Possible: Here’s How Kafka Does It. confluent.io, June 2019. Archived at perma.cc/Q2AU-Q2ED ↩︎

  79. Jason Gustafson, Flavio Junqueira, Apurva Mehta, Sriram Subramanian, and Guozhang Wang. KIP-98 – Exactly Once Delivery and Transactional Messaging. cwiki.apache.org, November 2016. Archived at perma.cc/95PT-RCTG ↩︎

  80. Pat Helland. Idempotence Is Not a Medical Condition. Communications of the ACM, volume 55, issue 5, page 56, May 2012. doi:10.1145/2160718.2160734 ↩︎

  81. Jay Kreps. Re: Trying to Achieve Deterministic Behavior on Recovery/Rewind. Email to samza-dev mailing list, September 2014. Archived at perma.cc/7DPD-GJNL ↩︎

  82. E. N. (Mootaz) Elnozahy, Lorenzo Alvisi, Yi-Min Wang, and David B. Johnson. A Survey of Rollback-Recovery Protocols in Message-Passing Systems. ACM Computing Surveys, volume 34, issue 3, pages 375–408, September 2002. doi:10.1145/568522.568525 ↩︎

  83. Adam Warski. Kafka Streams – How Does It Fit the Stream Processing Landscape? softwaremill.com, June 2016. Archived at perma.cc/WQ5Q-H2J2 ↩︎

  84. Stephan Ewen, Fabian Hueske, and Xiaowei Jiang. Batch as a Special Case of Streaming and Alibaba’s contribution of Blink. flink.apache.org, February 2019. Archived at perma.cc/A529-SKA9 ↩︎

最後更新於
11. 批處理13. 流式系統的哲學