Chapter 11. Stream Processing Engine: Toward Unified Data Processing

Ryan Fu

78 min read

·

May 29, 2026

--

Share

《給學弟妹的矽谷 AI / Software 指南》是一個矽谷工程師自發的眾籌寫作計劃，由源來適你社群推動。這將是一個在台灣軟體學生圈內很有影響力的文章系列。已有 >10 篇準備發布，目標 100 篇並出書，歡迎 tag 強者大神來寫！

Press enter or click to view image in full size

在 Stream Processing Engine 的世界裡，當資料以流的形式不斷湧入，系統必須在毫秒之間做出決策。任何一次重試，都可能造成重複計算；任何一次延遲，都可能改變結果的正確性。於是，三個看似獨立的問題 — — 精確一次（exactly-once）、狀態（state），以及時間（time） — — 開始交織在一起。

如果沒有狀態，我們無法記住過去；
如果沒有時間，我們無法判斷順序；
而如果兩者都不穩定，精確一次便無從談起。

於是，一個更根本的問題浮現：當我們同時面對這三個核心挑戰時，一個完整的 Stream Processing Engine，究竟是如何被建構出來的？

三個核心挑戰，並不是在同一時間被解決的。它們在不同的技術節點之間反覆拉扯，答案最終在 Apache Flink 成形。

2013 年，Google MillWheel 嘗試從工程層面出發，第一次正面迎戰精確一次與狀態的問題，並開始處理時間所帶來的實務挑戰。然而，在這個階段，時間仍然只是被使用的工具，而不是被理解的概念。MillWheel 已經具備 stateful processing 與 exactly-once 的雛形，並引入 low watermark 等機制，使系統能夠在現實環境中運作。這些設計，為後來 Flink 在 checkpoint、state 管理，以及容錯機制上的實作奠定了工程基礎，但仍缺乏一套完整的時間語意，使系統難以精確推理結果的正確性。

2015 年，Google Dataflow 提出了一個截然不同的觀點：時間不再只是輔助計算的參數，而是整個系統的核心語意。Event Time、Watermark 等概念，使我們第一次能夠在理論上定義「正確的結果」。Flink 後來的 event-time processing、watermark 機制與 window semantics，正是在這樣的思想下逐步建立起來，使 streaming 計算不再只是即時處理資料，而是能夠對結果的正確性進行推理與保證。

2016 年，Apache Beam 則將 Dataflow 轉化為一種可被撰寫、可被移植的程式模型，讓這些原本抽象的概念，成為工程師可以直接操作的語言。

而故事並沒有停在這裡。

受 Dataflow 啟發，接下來的 Flink Dynamic Tables 和《One SQL to Rule Them All》這篇論文讓整個資料世界開始重新理解：

• stream
• batch
• table
• SQL

這些抽象概念彼此之間的關係。

這段發生於 2013 到 2019 年之間的歷史，交織著工程底層實作、理論、模型與落地等不同面向，是一段人類逐漸學會在「不確定的時間流」中建立確定性的過程。

而在這段歷史之中，三次極其重要的「抽象融合」也悄悄發生。我將它們作為彩蛋，散落於整個篇章之中，歡迎讀者自行尋找。

MillWheel: The Battle for Correctness (2013)

Introduction

在 2010 年代初期，streaming 系統已經開始被廣泛應用。從 Apache S4 到 Apache Storm，工程師已經可以將資料建模為持續流動的事件，並透過分散式系統即時地進行處理。

然而，這些系統所解決的，多半是「如何讓資料持續流動」的問題。當資料可以被即時處理之後，一個更根本的挑戰開始浮現：在面對機器失敗、網路延遲與資料亂序的現實環境下，我們是否仍然能夠得到正確的結果？

MillWheel 正是在這樣的背景下誕生。作為 Google 內部用於大規模流式資料處理的系統，MillWheel 並不只是關注吞吐量或延遲，而是將焦點放在一致性（consistency）與正確性（correctness）上。

在其設計中，資料被建模為持續流動的 record，系統以一張分散式的資料流圖來組織計算，並支援 per-key 的狀態管理、細粒度的 checkpoint，以及對使用者透明的 exactly-once 處理語意。這些能力，使得 streaming 系統第一次能夠在實務上同時兼顧即時性與正確性。

也正因如此，MillWheel 並不是另一個「更快的 streaming 系統」，而是第一次嚴肅地回答了一個問題：當資料持續流動、且世界充滿不確定性時，我們如何仍然建立一個可被信任的計算系統？

Components

Computation

在 MillWheel 中，整個系統被建模為一張持續運作的資料流圖，而 Computation，便是這張圖中的基本節點。每一個 Computation 代表一段使用者定義的處理邏輯。它從一個或多個 stream 接收資料，經過內部的運算後，再將結果輸出到其他 stream，形成一條不斷流動的處理管線。

從結構上來看，一個 Computation 包含三個核心要素：輸入的資料流、輸出的資料流，以及節點內部的處理邏輯。這樣的設計，使 Computation 在概念上接近於 Apache Storm 中的 Bolt，或 Apache S4 中的 Processing Element。

然而，MillWheel 並不只是重現這些早期系統的設計，而是進一步將 state、fault tolerance 與一致性語意納入 Computation 的執行模型之中，使其成為一個可以支撐正確性保證的運算單位。

Stream

在這張資料流圖中，Stream 扮演的是節點之間的連結。它描述了資料從一個 Computation 流向另一個 Computation 的方向，也構成了整個系統的運作結構。一個 Computation 可以訂閱多個 stream，並同時將結果輸出到不同的 downstream 節點。這些 stream 並不是靜態的資料集合，而是持續流動的 record 序列，代表著系統中不斷發生的事件。

與傳統批次系統中「先儲存、再處理」的模式不同，Stream 中的資料在產生的同時就被消費與轉換。資料不再等待，而是持續流動。也正因如此，Stream 不只是資料的載體，它同時定義了整個系統的時間特性與處理節奏。

Key

如果說 Stream 定義了資料如何流動，那麼 Key 則決定了資料如何被理解與計算。在 MillWheel 中，每一筆資料都被表示為一個三元組：（key, value, timestamp）。其中，key 是用來分群與定位資料的核心維度。

與 Apache Storm 或 Apache S4 不同，MillWheel 將 key 的決定權交給資料的消費端。每一個 Computation 都可以針對輸入的 stream，自行定義 key 的萃取方式。這意味著，同一條資料流，可以被不同的 Computation 以不同的方式解讀與聚合，而不需要複製成多條幾乎相同的 stream。資料本身保持不變，但其意義會隨著不同的 key 被重新詮釋。

更重要的是，Key 同時也是計算的基本單位。系統會針對每一個 key 維護對應的 state，並保證同一個 key 的資料會依序處理，而不同 key 之間則可以平行執行。

在這樣的模型下，stateful computation 不再是額外的機制，而是自然地融入資料流之中。

Persistent State

如果說 Computation、Stream 與 Key 定義了資料如何流動與被分群，那麼 Persistent State，則決定了系統是否能夠「記住」過去。在 MillWheel 中，每一個 Computation 都可以為每一個 key 維護對應的狀態。這些狀態並不是暫存在記憶體中的中間結果，而是被持久化儲存（persistent），並與系統的容錯機制緊密結合。

具體來說，當一筆資料被處理時，系統會同時更新對應 key 的 state，並將這些變更納入 checkpoint。這些 checkpoint 是細粒度的，意味著系統不需要定期暫停整個資料流來進行快照，而是能夠在資料持續流動的過程中，逐步地記錄狀態的變化。

這樣的設計，使得在機器失敗或任務重啟時，系統可以從最近的狀態恢復，並避免重複計算所帶來的不一致問題。對使用者而言，這提供了一種近似 exactly-once 的處理語意 — — 即使底層發生重試，計算結果仍然保持一致。

更重要的是，state 並不是附加在系統之上的輔助功能，而是整個計算模型的一部分。每一個 Computation 的邏輯，都可以自然地建立在 state 之上，使得像是計數、聚合、去重等操作，成為 streaming 計算的基本能力，而不再需要額外的外部系統來支援。

Low Watermarks

然而，光有 state 並不足以解決 streaming 中的所有問題。當資料以非同步的方式到達系統時，一個更困難的問題隨之出現：我們如何知道「某個時間點之前的資料已經到齊」？

在 MillWheel 論文中，這個概念被稱為 Low Watermark，用來強調它代表的是 event time progress 的「下界估計（lower bound estimate）」。後來在 Dataflow 與 Flink 等系統中，社群逐漸簡稱為 watermark。

low watermark 可以被理解為系統對資料進度的一種估計：當 watermark 推進到某個 timestamp 時，系統便認為所有早於這個時間點的資料，已經被接收並處理完成。

這樣的機制，使得系統即使在面對 out-of-order data 時，仍然可以進行基於時間的推理。例如，一個 Computation 可以在 watermark 超過某個時間點後，對該時間區間的資料進行聚合，或輸出最終結果，而不需要等待所有資料嚴格依序到達。

然而，在 MillWheel 中，watermark 仍然是一種工程上的進度指標，而不是一套完整的時間語意模型。它提供了一種實務上可行的方式，讓系統能夠在不確定的資料到達順序中繼續前進，但並未嚴格定義在何種情況下結果應被視為最終且正確。

這也意味著，雖然 MillWheel 已經開始觸及「時間」這個問題，但對於時間如何影響計算結果，仍缺乏一套可以被形式化推理的框架。

Timers

如果說 Persistent State 讓系統能夠記住過去，Low Watermark 讓系統能夠推進時間，那麼 Timers，則讓系統開始對時間做出回應。

在 MillWheel 中，每一個 Computation 都可以為特定的 key 設定 timer。這些 timer 與 key 的 state 綁定，並以 timestamp 作為觸發條件。當系統的時間進度推進到指定的時間點時，對應的 timer 便會被觸發，進而執行使用者定義的邏輯。

這樣的機制，使得 Computation 不再只是被動地等待資料到達，而是可以在「某個時間點」主動執行操作。例如，在進行時間區間的聚合時，系統可以在某個時間點觸發計算，輸出該時間範圍內的統計結果，而不需要等待一個明確的結束訊號。

更重要的是，timer 的觸發與 state 的更新，是在同一個一致性模型之中運作的。也就是說，timer 所觸發的操作，會與資料處理一樣，被納入系統的容錯與重試機制中，使其在面對失敗時，仍然能夠保持結果的一致性。

然而，在 MillWheel 中，timer 仍然是一種以工程為導向的機制。它提供了一種讓系統在特定時間點執行邏輯的能力，但並未與 watermark、window 等概念形成一套完整的時間語意模型。換句話說，系統可以在「某個時間點」做出反應，但對於「這個時間點代表什麼」，以及「結果何時算是最終」，仍缺乏一套可以被清楚推理的框架。

為了理解 MillWheel 中資料究竟是如何流動的，我們可以用一個簡單的廣告點擊統計系統作為例子。

An Example: Counting Ad Clicks in Real Time

假設我們希望即時統計每一則廣告在每分鐘內的點擊次數，並在每分鐘結束後，將結果輸出到下游系統。

Press enter or click to view image in full size

Step 1: Event Enters the Stream

現在，系統收到一筆事件：

• (AdID = “AD_42, “UserID = “U1007”, EventTime = 2026–05–23 10:00:12)

這筆資料首先進入一條 Stream：

• click_stream

在 MillWheel 中，Stream 是 Computation 與 Computation 之間的資料流向。

Step 2: Computation Extracts the Key

接著，ClickCounter 這個 Computation 訂閱了這條 stream。Computation 定義為：

• key_extractor = AdID

因此，剛剛那筆事件會被解析為：

• Key = “AD_42”
• Value = click_event
• Timestamp = 2026–05–23 10:00:12

這意味著，所有屬於 AD_42 的事件，都會被路由到相同的邏輯 state 中。

Step 3: Persistent State Is Updated

此時，ClickCounter 會讀取對應 key 的 Persistent State：

• state[“AD_42”][“10:00”]

假設目前 state 中的值是：

• count = 57

收到新事件後：

• count = 58

這個更新不會只存在記憶體中，而是會被持久化儲存。即使 worker 在下一秒失敗，系統仍然可以從 persistent state 恢復計算。論文中提到，MillWheel 的持久化能力建立在 Google 內部的 distributed storage infrastructure 之上，其中包含 Bigtable 等系統。

Step 4: Low Watermark Advances

隨著更多事件持續到達，這條 stream 在這個 Computation 中的 Low Watermark，也會逐漸向前推進。

假設目前這個 Computation 已經觀察到以下事件：

• 10:01:05
• 10:01:08
• 10:01:10

系統目前所觀察到的最大 event timestamp 為：

• 10:01:10

然而，MillWheel 並不會立刻將 watermark 推進到 10:01:10。因為系統知道，在分散式環境中，資料可能仍然因為網路延遲、partition skew，或 upstream backpressure 而晚到。

假設系統根據歷史統計，認為：

• 大部分事件的延遲不會超過 10 秒

那麼它可能會保留：

• 10 秒 safety margin

於是，這個 Computation 的 Low Watermark 便會被推進到：

• 2026–05–23 10:01:00

這代表系統認為：

• 未來已經不太可能再收到 timestamp 小於 10:01:00 的事件。

注意，Low Watermark 並不是資料本身攜帶的欄位，也不是整個系統共享的一個全域時間。它更像是系統對 event-time progress 的動態估計。

在 MillWheel 中，watermark 會隨著 stream 在不同 Computation 之間逐步傳遞。每個 Computation 都會根據自己所接收到的 input stream，維護對應的 watermark 狀態。

如果某個 Computation 同時依賴兩條 input streams：

• stream_A watermark = 10:01:00
• stream_B watermark = 10:00:40

那麼實際的 input watermark，為了等待最慢的那條 stream，通常會取兩者中的較小值：

• min(10:01:00, 10:00:40) = 10:00:40

而 watermark 的推進速度，本質上是一種 correctness 與 latency 之間的 trade-off。例如：

• safety margin = 2 秒，watermark = 10:01:08，結果 latency 很低，但 late data 風險高
• safety margin = 30 秒，watermark = 10:00:40，結果 correctness 較高，但 latency 增加

除了 safety margin 之外，watermark 的行為也會受到其他因素影響。例如：

• allowed lateness estimate：系統若允許 late data <= 1 minute，則 watermark 通常會更保守。
• source delay model：若 source 位於跨洲資料中心，系統可能預期 network delay ≈ 15 秒，則 watermark 推進也會較慢。
• out-of-order 程度：若事件經常亂序，則系統通常需要保留更大的 safety margin。
• propagation policy：某些系統可能每 100 ms 傳播一次 watermark；有些則可能每 5 秒才更新一次 downstream watermark。這會影響 timer 觸發與 window completion 的即時性。
• watermark 更新頻率：即使 watermark estimate 本身相同，若系統每 10 秒才更新一次 watermark，下游仍然會感受到額外延遲。

總結來說，Low Watermark 是系統對時間的信心。

Step 5: Timer Fires

然而，僅僅有 state 與 watermark，系統仍然不知道「什麼時候該輸出結果」。因此，在一開始處理資料時，這個 Computation 便已經為對應的時間窗口註冊了一個 timer：

• window = [10:00:00, 10:01:00)
• fire_time = 10:01:00

隨著前一節中的 Low Watermark 持續向前推進：

• watermark = 10:01:00

這個 timer 終於被觸發（fire）。此時，Computation 會讀取對應 key 的 Persistent State：

• state[“AD_42”][“10:00”] = 58

並產生一筆輸出 record：

• (“AD_42”, “10:00”, 58)

這筆 output record 會被寫入 downstream stream，並傳遞給下一個 Computation。後續系統可能再將其聚合、寫入分析儲存，或更新即時 dashboard。

在這個例子中，它代表：

廣告 AD_42 在 10:00 ~ 10:01 之間，共收到 58 次點擊。

在 MillWheel 中，Timer 並不是單純依賴 wall clock 的鬧鐘，而是與 event-time progress 緊密耦合。真正決定 timer 是否觸發的，不是機器現在幾點，而是 watermark 是否已經推進到對應時間。

這種設計非常關鍵。因為在 streaming 世界裡，資料往往會亂序到達。如果系統只依賴 processing time，許多較晚到達、但 event time 屬於舊 window 的資料，便可能被錯誤地排除。

透過 watermark 與 timer 的結合，MillWheel 第一次讓系統能夠：

• 等待「大部分應該到達的資料」
• 同時避免無限期等待
• 並在 correctness 與 latency 之間取得平衡

Step 6: Delivery Guarantees and Strong Productions

當 timer 被觸發後，系統終於準備將結果送往下游。然而，在 MillWheel 中，「計算完成」並不代表「結果已經安全」。

假設此時系統產生了以下 output record：

• (“AD_42”, “10:00”, 58)

這筆資料接著準備被寫入 downstream stream。但如果 worker 在送出過程中突然失敗，例如：

• at 10:01:02, worker crash

此時，系統可能陷入一個危險狀態：

• output 已經送出
• 但 upstream 尚未收到 ACK
• recovery node 不知道 downstream 是否已經收到資料

如果系統直接重新計算並再次送出 output，就可能產生 duplicate result；但如果選擇不重送，又可能造成資料遺失。

為了解決這個問題，MillWheel 論文引入了 Strong Productions。在這個模型中，output 並不會在計算完成後立刻送出，而是會先被 checkpoint 到 persistent backing store。

• checkpoint: (“AD_42”, “10:00”, 58)

只有當 checkpoint 完成之後，系統才真正將 output 發送到 downstream stream。這意味著 downstream output 本身，也成為系統 state 的一部分。因此，即使 worker 在 transmission 過程中失敗，新的 worker 也不需要重新執行 window aggregation，而是可以直接 replay 已經 checkpoint 的 output。這正是 Strong Productions 的核心：系統保證「輸出的事實」可以被重建。

相對地，如果系統沒有先 checkpoint output，而只是直接將資料送往下游，那便屬於一種較弱的 delivery model，可以被視為 Weak Productions。

在 Weak Productions 中：

• output 不一定可 replay
• failure 後可能需要重新計算
• downstream 可能看到 duplicate outputs
• correctness 更依賴 application-level deduplication

這也是為什麼 MillWheel 能夠逐步逼近 exactly-once semantics，而早期 stream processing engine 往往只能做到 at-least-once。

而 checkpoint 本身，也與 time window 有著微妙但重要的關係。例如可以這樣設置：

• window = 1 minute
• checkpoint interval = 5 seconds

window 決定的是「哪些事件屬於同一組計算」，而 checkpoint interval 決定的則是「系統多久保存一次目前進度」

較短的 checkpoint interval：

• recovery 更快
• duplicate risk 更低
• 但 storage 與 I/O 成本更高

較長的 checkpoint interval：

• overhead 較低
• 但 failure 後可能需要 replay 更多 state changes

因此，checkpoint 本質上並不是時間語意的一部分，而是 fault tolerance 的工程機制。

Dataflow: Rethinking Time in Data (2015)

Introduction

MillWheel 幾乎解決了 streaming 系統中所有工程層面的困難。

資料可以被可靠地處理，狀態可以被持久化，失敗可以被重試，輸出也可以透過 Strong Productions 保證一致性。Low Watermark 與 event-time-aware timer 的引入，更讓系統第一次真正開始理解 event time，而不只是單純依賴 processing time。

從工程的角度來看，一個完整的 stream processing engine，似乎終於開始成形。

然而，Dataflow 所做的，並不是再發明另一個新的 streaming system，而是進一步將這些原本分散在系統中的概念，抽象成一套完整的時間語意模型（Time Semantics）。此後，時間不再只是 timer 背後的隱藏機制，也不只是系統內部對 progress 的估計，而成為開發者可以直接描述與操作的計算語意。

Time Semantics

Press enter or click to view image in full size

Dataflow 的 Time Semantics，將原本隱含在系統中的「時間」，拆解成一組可以被明確描述的語意，可以同時回答四個問題：

• 我們在計算什麼（What）
• 在哪個時間範圍內計算（Where）
• 什麼時候輸出結果（When）
• 結果如何隨時間修正（How）

這四個維度，構成了 Google Dataflow 對 streaming 計算的完整描述。

過去的 streaming 系統，往往將時間視為系統內部的實作細節；而 Dataflow，則第一次將時間提升為一級語意（first-class semantics）。時間不再只是 timer 或 scheduler 背後的隱藏機制，而成為開發者可以直接描述與操作的計算模型。

值得注意的是，Dataflow paper 本身仍然偏向概念與語意模型的建立；真正將這些抽象概念轉化為工程 API 的，則是後來的 Apache Beam。

因此，接下來在介紹 What、Where、When 與 How 這四個維度時，我除了會說明 Dataflow paper 中的核心思想，也會同時搭配 Beam 的 API 與 programming model，讓這些原本抽象的時間語意，能夠更具體地被理解。

Transform: Defining What

在 Dataflow 中，所有資料處理都建立在兩個核心 primitive 之上：

• ParDo
• GroupByKey

而這些 primitive，在 Dataflow 裡被統稱為 Transform，是一種描述「資料應該如何被轉換」的計算步驟。它可以是：

• 對每筆事件進行處理
• 對資料重新分組
• 對多筆資料進行 aggregation
• 對資料進行 filter、join 或 combine

整個 Pipeline，本質上就是由一連串 Transform 所構成的計算圖（computation graph）。其中，ParDo 負責描述逐筆事件的轉換邏輯，例如：

• parse event
• extract fields
• filter invalid records

而 GroupByKey，則負責將具有相同 key 的資料聚合在一起。例如：

• group clicks by ad_id

以下就是在 Apache Beam API 的體現：

# ParDo
events | beam.Map(parse_event)

# GroupByKey
events
| beam.Map(lambda x: (x.ad_id, 1))
| beam.GroupByKey()

這些操作並不是新的概念，MapReduce 裡就有。現在我們要處理的棘手問題在於：當資料是無界、亂序且持續到達時，GroupByKey 應該在什麼時候被視為完成？

在 batch 世界裡，這個問題並不存在。系統只需要等待所有資料到齊，再進行 aggregation 即可。但在 streaming 世界中，「所有資料」本身可能永遠不存在。

正因如此，我們需要定義 Where。

Windowing: Defining Where

Windowing 回答的是：我們應該在哪個時間範圍內進行計算？

在無界資料流中，我們無法等待「所有資料」到齊，因此必須主動切分資料。Windowing 提供了一種方式，將連續不斷的事件，劃分為一段一段有限的時間區間。

這些區間可以是固定大小的時間窗口（Fixed Window），例如：

• 每 1 分鐘統計一次廣告點擊數

也可以是滑動窗口（Sliding Window）：

• 每 5 分鐘統計一次，但每 1 分鐘更新結果

甚至可以是依據事件分佈動態形成的 Session Window。例如：

• 若使用者 30 分鐘內沒有新的操作，則視為 session 結束

透過這樣的切分，原本無界的資料流，得以在邏輯上被轉換為一系列有限集合，使得像 GroupByKey 這樣原本只能在 batch 系統中完成的操作，也能重新在 streaming 世界中成立。

以下是三種 windows 在 Beam API 的實現：

# 60 sec Fixed Window
beam.WindowInto(
   beam.window.FixedWindows(60)
)

# Sliding Window
# 300 sec window, slide every 60 sec
beam.WindowInto(
   beam.window.SlidingWindows(
       size=300,
       period=60
   )
)

# Session Window
# 1800 sec inactivity gap
beam.WindowInto(
   beam.window.Sessions(1800)
)

除此之外，Beam 也允許開發者自訂 WindowFn，建立自己的 window assignment logic。

值得注意的是，Window 並不一定只能建立在 event time 之上。在 Dataflow 與 Beam 的模型中，Windowing strategy 與 time domain，其實是兩個彼此獨立的維度。最常見的時間語意包括：

• Event Time
• Processing Time
• Ingestion Time

其中：

• Event Time 指事件真正發生的時間
• Processing Time 指 worker 實際觀察到事件的時間
• Ingestion Time 則是資料進入 streaming infrastructure 的時間

但更深一層來看，Dataflow 的時間模型其實並不限於 wall clock。理論上，任何能夠描述「系統進度（progress）」的 logical coordinate，都可以成為時間語意的一部分。例如：

• Database CDC 的 SCN / LSN / binlog offset
• ML training 中的 epoch / global step
• Trading system 中的 market tick sequence
• distributed system 中的 logical timestamp

Dataflow 真正革命性的地方在於，它開始暗示「時間」不一定來自時鐘，也可以源於任何描述系統進度的座標。

而當系統開始採用 event time 時，便需要 Watermark 這種 heuristic progress estimation，也就是系統對於「資料完整度」的推測。在 Beam 中，開發者可以直接定義 watermark strategy。例如：

# watermark = max_event_time_seen - 30 sec
WatermarkStrategy
    .forBoundedOutOfOrderness(
        Duration.ofSeconds(30)
)

而更進一步地，Beam 與 Flink 也允許開發者自訂 watermark generator。從此，watermark 不再只是 runtime 內部的隱藏機制，而開始成為可被工程師直接操作的 programming abstraction。

理論上，不同的：

• window strategy
• time domain
• watermark policy

可以形成大量組合。例如：

• Fixed Window + Event Time + (max_event_time — delay)
• Sliding Window + Event Time + bounded lateness
• Session Window + Event Time + heuristic watermark
• Fixed Window + Processing Time + system clock

但並不是所有組合都有道理，例如以下設置或許無法跟老闆交代：

• Session Window + Processing Time

因為 Session 本質上是在描述 user behavioral continuity，所以更適合建立在 Event Time 之上。

決定資料該如何分組計算後，下一步須要定義 When。

Triggering: Defining When

Trigger 回答的是：我們應該在什麼時候輸出結果？

在現實世界中，資料可能延遲、亂序，甚至在很久之後才到達。Trigger 的引入，讓輸出從一次性的事件，轉變為一個可被控制的過程。

在 Beam 中，default trigger 會在 watermark 判斷某個 window 的資料大致到齊時，輸出該 window 的結果；如果沒有額外設定 late trigger，後續 late data 可能被丟棄。Beam 文件也明確說明，trigger 的作用就是決定每個 window 何時 emit aggregation result，這些被輸出的結果稱為 pane。

最基本的，是 Event-Time Trigger。當 watermark 超過 window 結束時間時，系統便認為大部分屬於這個 window 的資料，應該都已經到達。

例如：

• window = [10:00, 10:01)
• watermark = 10:01

此時，系統便可以輸出：

• AD_42 = 58 clicks

在 Beam Python API 中，可以寫成：

from apache_beam import window
from apache_beam.transforms import trigger

events | beam.WindowInto(
    window.FixedWindows(60),
    trigger=trigger.AfterWatermark()
)

然而，watermark 永遠只是估計，而不是絕對保證。因此，Dataflow 也允許 Late Trigger 的存在。假設在 watermark 已經推進之後，又有一筆較晚到達的事件：

• Another event with timestamp = 10:00:42
• arrival time = 10:03:15

系統仍然可以重新觸發 trigger，輸出更新後的結果：

• AD_42 = 59 clicks

在 Beam 中，可以把 late trigger 接在 AfterWatermark 後面：

# watermark 到達後先輸出一次；若 5 分鐘內還有 late data 到達，每來一筆 late data，就可以再次輸出更新結果。
events | beam.WindowInto(
    window.FixedWindows(60),
    trigger=trigger.AfterWatermark(
        late=trigger.AfterCount(1)
    ),
    allowed_lateness=300
)

除了 event-time-based trigger 之外，Dataflow 論文也提到 Processing-Time Trigger。它關心的不是 event time 是否完整，而是希望降低使用者看到結果的 latency。例如：

• 每 10 秒先輸出一次目前結果

在 Beam 中，可以寫成：

events | beam.WindowInto(
    window.FixedWindows(60),
    trigger=trigger.AfterProcessingTime(10)
)

此外，系統也可以使用 Data-Driven Trigger，在高流量系統中，這種方式可以避免長時間沒有輸出。例如：

• 每累積 1000 筆事件就輸出一次

在 Beam 中，可以寫成：

events | beam.WindowInto(
    window.FixedWindows(60),
    trigger=trigger.AfterCount(1000)
)

更重要的是，這些 Trigger 並不是互斥的。Beam 支援把 event-time、processing-time、data-driven trigger 組合起來。AfterWatermark 本身也支援 early 與 late trigger 參數：early 用來在 watermark 到達前提早輸出，late 用來在 watermark 到達後處理 late data。例如：

# 先每 10 秒輸出一次近似值, watermark 到達後輸出正式結果, late data 到達時再重新修正
events | beam.WindowInto(
    window.FixedWindows(60),
    trigger=trigger.AfterWatermark(
        early=trigger.AfterProcessingTime(10),
        late=trigger.AfterCount(1)
    ),
    allowed_lateness=300
)

不同 trigger 的選擇，本質上是在 trade-off：

• 越早輸出：latency 越低，但結果越可能不完整
• 越晚輸出：correctness 越高，但使用者等待越久
• 越常輸出：互動性越好，但 cost 與 downstream 負擔越高

這也代表我們不再需要等待一個「完美時刻」才輸出結果；而是可以先提供較快但較不完整的答案，再逐步逼近更正確的結果。而 correctness、latency 與 cost，也第一次被放到同一個模型中共同權衡。

決定好該在何時輸出結果後，下一步必須決定 How。

Accumulation: Defining How

在傳統 batch 系統中，結果通常只有一次輸出，因此「修正結果」這件事幾乎不存在。但在 Dataflow 中，Trigger 可以讓同一個 window 被多次觸發，系統可能產生：58 clicks, 59 clicks, 61 clicks … 這些同屬一個 window 的 pane。

因此，系統必須進一步定義：

• 新結果應該覆蓋舊結果？還是累積在舊結果之上？是否要撤回舊結果？

論文中，主要描述了三種核心 accumulation mode。

Discarding Mode

每次 trigger 之後，系統都會丟棄先前已經輸出的 accumulation state。例如：

第一次 trigger：

• 58 clicks

之後又有 late data 到達：

• +1 click

下一次 trigger 時，系統只輸出：

• 1 click

每次 output，只代表「新增變化（delta）」。在 Beam 中，可以寫成：

beam.WindowInto(
    beam.window.FixedWindows(60),
    trigger=trigger.AfterWatermark(
        late=trigger.AfterCount(1)
    ),
    accumulation_mode=
        trigger.AccumulationMode.DISCARDING
)

在 Discarding Mode 中，trigger fire 之後，先前 pane 的 accumulation state 可以被丟棄。系統不需要長期保留 58 clicks 這類舊 aggregation state，而只需要處理新的變化量。因此：

• retained state 較小
• recomputation 成本較低
• network traffic 較低

這種模式通常較適合：

• high-throughput pipeline
• append-only downstream
• monitoring system
• 對 storage cost 敏感的場景

但相對地：

• downstream 必須自行累積結果
• consumer 必須理解 pane 是 delta
• downstream logic 會變得較複雜

Accumulating Mode

而在 Accumulating Mode 中，系統則會保留先前的 accumulation state。

第一次 trigger：

• 58 clicks

late data 到達後：

• +1 click

下一次 trigger：

• 59 clicks

每一次 output，都代表目前系統所相信的「最新真相」。在 Beam 中，可以寫成：

beam.WindowInto(
    beam.window.FixedWindows(60),
    trigger=trigger.AfterWatermark(
        late=trigger.AfterCount(1)
    ),
    accumulation_mode=
        trigger.AccumulationMode.ACCUMULATING
)

真正重要的是，系統必須保留完整 accumulation state。因為 59 clicks 並不是憑空產生的。系統必須記得 previous aggregation = 58，才能重新計算新的結果。而在其他 streaming 用例中，保留的 state 可能包括：

• Top-K aggregation
• Join state
• distinct set
• session state
• partial aggregation hash table

因此，Accumulating Mode 往往意味著：

• 更大的 retained state
• 更高的 recomputation cost
• 更大的 output volume
• 更高的 network traffic

但相對地，它對 downstream system 也更加直觀。因為 downstream 不需要自行維護 state，每一次 output，都可以直接被視為 current truth so far。因此，Accumulating Mode 通常較適合：

• dashboard
• BI system
• streaming SQL
• materialized view
• user-facing analytics

總結來說，Discarding 與 Accumulating 的選擇一種 trade-off：

• Discarding：較低 cost，但對 downstream 來說較複雜
• Accumulating：較高 cost，但對 downstream 來說較直觀

Accumulating and Retracting Mode

前面兩種模式，對於像 click count 這種單純累加的 aggregation 已經足夠。但有些 query 的結果並不是「只會增加」的，最典型的例子是 Top-K。

假設我們正在計算：

• 每分鐘點擊數最高的前 3 個廣告

一開始，window [10:00, 10:01) 的結果是：

• Top 3 = A, B, C

如果後來一筆 late data 到達，使得廣告 D 的點擊數超過 C，新的結果就變成：

• Top 3 = A, B, D

這時候，問題不只是「多了一個 D」，而是「C 必須被移出結果」。

Discarding Mode 只能輸出新的 delta，但無法清楚表達 C 已經失效。Accumulating Mode 可以輸出新的完整結果 A, B, D，但如果 downstream 不是覆蓋式儲存，而是把每次輸出都當成新的 row，就可能同時保留 A, B, C 與 A, B, D，造成語意混亂。

因此，這類 query 需要第三種模式：Accumulating and Retracting。它不只輸出新的 refinement，也明確撤回先前已經不再正確的結果。

需要注意的是，Beam 還沒實作 Retraction API，因此這次我們以 Flink SQL 來舉例 explicit retraction。假設我們寫：

SELECT *
FROM (
 SELECT
   ad_id,
   COUNT(*) AS clicks,
   ROW_NUMBER() OVER (
     ORDER BY COUNT(*) DESC
   ) AS rank_num
 FROM clicks
 GROUP BY ad_id
)
WHERE rank_num <= 3;

一開始，系統得到：

A = 100 clicks
B = 90 clicks
C = 80 clicks

因此輸出這些 changelog：

INSERT (rank=1, A, 100)
INSERT (rank=2, B, 90)
INSERT (rank=3, C, 80)

接著，一筆 late data 讓 D 的點擊數變成 85：

D = 85 clicks

新的 Top 3 變成：

A = 100
B = 90
D = 85

此時，Flink 不能只是輸出 D。它還必須告訴 downstream：原本排名第三的 C 已經不屬於結果。因此 changelog 就會變成：

DELETE / RETRACT (rank=3, C, 80)
INSERT (rank=3, D, 85)

如果排名也發生變動，例如 D 超過 B，則系統還需要撤回舊排名，再輸出新排名：

UPDATE_BEFORE (rank=2, B, 90)
UPDATE_BEFORE (rank=3, C, 80)
UPDATE_AFTER (rank=2, D, 95)
UPDATE_AFTER (rank=3, B, 90)

這就是 Accumulating and Retracting 的核心精神：系統一方面保留並更新目前最完整的結果，另一方面也明確撤回那些已經不再成立的舊結果。也因此，Top-K、Rank、Join、Distinct 這類 non-monotonic query，往往不能只靠單純累加來處理。它們需要一種能夠表達「舊結果失效」的語意，而這正是 Retraction 存在的理由。

Unifying Batch and Streaming

在傳統觀點中，batch 處理的是一個已經完成的資料集合，而 streaming 則處理持續到達的資料流。兩者在系統設計與實作上，看似有著本質上的差異。

如果我們將資料視為一個無界的資料流，那麼所謂的 batch，其實只是這個資料流在某一個時間範圍內的切片。換句話說，batch 並不是一種不同的資料型態，而是 streaming 在特定條件下的一種觀察方式。

而這樣的統一，正是透過：

• Window
• Trigger
• Accumulation

這三個時間語意所完成的。從這個角度來看，streaming 與 batch 的差異，在於對這三種語意採取了不同策略。

在典型的 batch 系統中：

• Window 通常是一個具有明確邊界的完整資料集，例如：
  2026–05–01 ~ 2026–05–31 的所有資料
• Trigger 通常只會觸發一次，也就是：
  等待所有資料完成後才輸出
• Accumulation 則通常採取：
  final immutable result，一旦輸出，結果便不再改變。

因此，在 batch 的世界裡，時間被隱藏了，系統假設資料終究會完整到齊。而在 streaming 系統中，這三者則採取完全不同的策略。

• Window 變成持續移動、持續切分的時間範圍。
• Trigger 不再只有一次，而可能：提前輸出近似值、watermark 到達時輸出正式結果、late data 到達後再次修正
• Accumulation 也不再是 immutable final answer，而是一個 continuously refining result

在這樣的模型裡，streaming 不再只是「即時版 batch」，反而成為更一般化、更接近真實世界的資料模型；而 batch，則只是 streaming 在資料已經靜止之後的一種觀察方式。這也意味著，我們不再需要為 batch 與 streaming 分別設計兩套完全不同的系統，而是可以在同一個時間語意模型之下，描述所有資料處理行為。

中國宋代詩人蘇軾曾寫下：『橫看成嶺側成峰，遠近高低各不同。』

而在西方物理學中，Quantum Mechanics 則揭示了另一種雙重性：粒子與波，並不是兩種不同的存在，而是同一個系統在不同尺度下的兩種表現。

batch 與 streaming 的關係，亦是如此。

Apache Beam: A Unified Programming Model (2016)

Press enter or click to view image in full size

From Model to Language

在 Dataflow 中，我們已經看見了一個關鍵轉變：資料處理不再只是操作資料，而是操作「時間中的資料」。Window、Trigger 與 Accumulation，讓我們第一次能夠精確描述在不完整世界中如何逐步逼近正確結果。

然而，這些概念該如何被工程師真正使用？

Apache Beam 的出現，正是為了回答這個問題。但要理解 Beam，我們必須先理解它的來源。

Beam 並不是一個憑空誕生的新系統，而是 Google 在建立 Dataflow 模型與 Cloud Dataflow 服務過程中的自然延伸。Dataflow 論文的作者群，同時也是這套系統的主要設計者。他們在內部實踐了這套模型之後，進一步將其抽象為一個通用的程式模型，並透過 Apache 基金會開源，最終形成了 Beam。

Press enter or click to view image in full size

換句話說，Beam 並不是另一種資料處理理論，而是 Dataflow 的語言化版本。在這個語言之中，資料被抽象為 PCollection — — 一個可能是有限（bounded），也可能是無限（unbounded）的資料集合。所有的操作，則被表達為一系列 Transform，從最基本的 ParDo，到涉及跨資料關聯的 GroupByKey，再到描述時間語意的 Window 與 Trigger。

這樣的設計，帶來了一個深遠的改變：時間語意不再隱藏在系統內部，而是成為程式的一部分。工程師不再只是撰寫「要做什麼」，而是同時描述「在哪個時間範圍內做」、「什麼時候輸出」，以及「結果如何隨時間演化」。

當這些元素被串接在一起時，便形成了一條 Pipeline。

這樣的 Pipeline 並不是單純的執行流程，而是一種宣告式的描述。開發者所撰寫的，並不是「如何逐步執行」，而是「資料應該如何流動與被轉換」。實際的執行方式，則被延後到更底層的系統中決定。

這個設計帶來了一個關鍵改變：資料處理程式，第一次可以被視為一個與執行引擎解耦的邏輯結構。Pipeline 描述的是「做什麼」，而不是「怎麼做」。

Write Once, Run Anywhere

在 Apache Beam 中，Pipeline 並不是直接被執行的，而是交由一個稱為 Runner 的元件來運行。不同的 Runner，對應不同的執行引擎，例如：

• Flink
• Spark
• Google Cloud Dataflow

理論上，同一份 Pipeline 可以在不同 Runner 上執行，而不需要修改程式碼。這樣的設計，提出了一個極具吸引力的承諾：Write Once, Run Anywhere。開發者不再需要為不同系統撰寫不同版本的資料處理邏輯，而是可以專注於描述：

• 資料如何被轉換
• 時間如何被定義
• 結果如何被演化

至於實際的執行方式，則交由底層 engine 負責。而這背後，其實隱含著更大的野心。Beam 試圖將資料處理本身，抽象成一個獨立於執行引擎之外的通用層：

• PCollection 描述資料
• Transform 描述運算
• Window 與 Trigger 描述時間語意

某種程度上，這與編譯器世界中的 IR（Intermediate Representation）非常相似。程式不再直接綁定某一個 runtime，而是先被轉化為一種更高層的通用表示，再交由不同 backend 執行。

Did Beam Win?

Apache Beam 描繪出了 streaming 世界最完整的藍圖。然而，Beam 有像 Kubernetes 那樣統治世界嗎？答案其實非常耐人尋味。

AWS (~30% Market Share)

目前全球雲端市場市佔率最高的，仍然是 Amazon Web Services，長期維持在約 30% 左右。然而，在 AWS 的 streaming ecosystem 中，Beam 從未成為主流 abstraction。

雖然理論上 AWS 可以透過 Runner 支援 Beam，但實務上，多數團隊仍然直接使用：

• Apache Flink
• Apache Spark

工程師通常直接撰寫：

• Flink API
• Spark Structured Streaming API

而不是先透過 Beam abstraction，再交由 Runner 執行。原因其實非常現實。在 production 環境中，開發者最終仍然需要理解：

• checkpointing
• backpressure
• state backend
• task scheduling
• memory tuning

這些 execution engine 的細節。而 Beam 的 abstraction，並沒有真正消除這些複雜度，只是將它們部分隱藏起來。

Azure (~20–25% Market Share)

第二大的雲廠商，則是 Microsoft Azure，市佔率大約落在 20% 到 25% 之間。Azure 同樣支援 Beam Runner，但 Beam 也從未成為其主要 streaming API。在 Azure ecosystem 中，更主流的仍然是：

• Spark Structured Streaming
• Databricks
• Apache Flink

特別是在 Databricks 與 Spark 生態逐漸壯大之後，Spark Structured Streaming 幾乎成為 Azure data platform 的事實標準。因此，大部分工程團隊仍然傾向直接操作 execution engine，而不是建立在 Beam abstraction 之上。

Google Cloud (~10–12% Market Share)

真正將 Beam 視為核心 abstraction 的，其實只有市佔率大約為 10% 到 12% 的 Google Cloud。Google Cloud Dataflow 幾乎就是 Beam 的原生執行平台。Beam 中的：

• Pipeline
• PCollection
• Window
• Trigger

都能直接映射到 Dataflow runtime。

某種程度上，Beam 可以被視為 Google 將 Dataflow paper 工程化後對外公開的 programming model。在 Google Cloud 中，Beam 並不是附屬 API，而是整個 streaming abstraction 的核心。

Abstraction vs Execution

Beam 至今仍沒能成為主流，有兩個深層原因。

首先，Beam 出現得太早。在 2016 年，大部分工程團隊甚至還沒有真正理解 event time、watermark 與 late data 的重要性。相比之下，直接操作 Spark 或 Flink，往往更加直觀，也更容易最佳化。

更根本的是，開發者最終選擇的，往往不是最優雅的 abstraction，而是最可控的 execution engine。Beam 所描繪的 abstraction，或許是 streaming 世界中最優雅的一套理論模型之一；但真實世界的資料系統，最終仍然充滿了 backpressure、state migration、checkpoint tuning、memory management、skew、deployment 與 failure recovery 這些無法被完全抽象化的工程細節。

當不同 Runner 的 semantics、performance、checkpoint behavior 與 state management 開始出現差異時，”Write Once, Run Anywhere” 便逐漸變成 “Write Once, Debug Everywhere”。

於是，當今世界採納的並不是 Beam 的 abstraction layer 本身，而是 Dataflow 所提出的 What、Where、When、How 這套對時間、狀態與結果演化方式的理解。

Flink Dynamic Tables: Relationalizing Streaming

Press enter or click to view image in full size

Apache Flink

Apache Flink 最早起源於德國柏林 Technical University of Berlin 的研究計畫 Stratosphere。2014 年，Stratosphere 正式捐贈給 Apache 基金會，並改名為 Apache Flink。與當時許多仍然建立在 batch 思維上的系統不同，Flink 從一開始就選擇了 streaming-first architecture，視 batch 為 bounded stream。

因此，Flink 早期的發展重心，主要集中在 event-time processing、watermark、stateful streaming，以及 exactly-once checkpoint 等 execution engine 層面的問題。

Dataflow 對 refinement semantics 的重新定義，已經開始動搖傳統 batch result 的靜態觀點。而當 streaming system 逐漸需要面對 SQL、Materialized View 與 incremental maintenance 等 relational world 的需求時，streaming 與 relational algebra 的融合，幾乎成為一種必然。

於是，大約從 2017 年開始，Flink 逐漸將重心擴展到 Table API、Streaming SQL 與 Dynamic Tables 等 higher-level abstraction。而這段演化過程中，Blink 原本是一個由中國團隊建立的 Flink fork，主要專注於 Streaming SQL、query optimizer、relational execution，以及更完整的 Table API。相比當時較偏 execution engine 的原始 Flink，Blink 更加重視 relational abstraction on streaming systems。後來，Blink 的大量成果逐漸被合併回 Apache Flink 主線之中。

到了 2019 年前後，特別是 Flink 1.9，Blink planner 開始成為官方的重要方向，而 Dynamic Tables 與 Streaming SQL 的整體模型，也在這段期間逐漸成熟。這也標誌著一個重要轉變：streaming system 不再只是處理 event 的 engine，而開始成為一個 continuously evolving relational system。

Materialized View and Incremental Maintenance

Dynamic Tables 並不是全新的東西，它更像是將傳統資料庫中的 Materialized View，推進到 unbounded streaming world 後的結果。

在傳統資料庫中，Materialized View（MV）本質上是一種預先計算並持久化儲存的 query result。例如：

SELECT ad_id, COUNT(*)
FROM clicks
GROUP BY ad_id;

系統不需要每次 query 都重新掃描整張 table，而是可以直接讀取已經預先計算好的結果。

然而，當 underlying table 發生變化時，系統是否需要重新計算整張 view？於是，資料庫世界開始發展出 Incremental View Maintenance (IVM)，也就是只根據資料的變化量（delta），更新 view 的結果。

例如，一筆新的 click event 到達後，系統不需要重新計算所有廣告的點擊數，而只需要將對應 ad_id 的 count 加一即可。

而當 streaming system 開始引入：

• stateful processing
• event time
• continuous aggregation
• refinement semantics

時，事情開始變得非常有趣。因為 streaming system 其實就是一個永遠不會停止的 incremental update system。在這個觀點下，Dynamic Tables 可以被理解為 continuously maintained materialized views。table 不再是靜態資料集，而是一個持續變化的 relational state；query result 也不再是一次性計算完成的 final answer，而是一個會隨時間不斷被更新的 dynamic view。

Flink Dynamic Tables 所做的事情，其實就是將傳統資料庫中的 Materialized View 與 Incremental View Maintenance，推進到 unbounded streaming world。

Changelog Stream

在 Dynamic Tables 的世界中，stream 被重新理解為 changelog of a table。每一筆 stream event，不再只是 moving data，而是：

• INSERT
• UPDATE_BEFORE
• UPDATE_AFTER
• DELETE

而這也形成了一個後來非常重要的觀點：Stream-Table Duality。在這個觀點下：

• stream 可以被視為 table 的 changelog
• table 則可以被視為 stream 在某個時間點的 materialization

值得注意的是，changelog stream 並不等於完整 relational algebra 本身，它更像是 relational result state 的變化紀錄。這些 relational algebra

• projection
• selection
• join
• aggregation
• ranking

仍然存在於 query execution 與 optimizer 層中。

changelog 跟 Write-Ahead Log (WAL) 兩者都在記錄 state change，但主要差異在於：

• WAL 描述的是 storage state 的變化
• changelog stream 描述的則是 relational state 的變化

One SQL to Rule Them All (2019)

Introduction

承接前一節 Flink Dynamic Tables 的發展，下一個自然浮現的問題是：如果 stream 與 table 已經可以被統一理解，那麼 SQL 本身，是否也能同時描述 batch 與 streaming？這正是《One SQL to Rule Them All》試圖回答的問題。

這篇 paper 的核心觀點其實非常大膽：如果 SQL 已經是資料世界最通用的語言，那麼 streaming 為什麼還需要另一套語言？作者的答案是：不需要。

而這篇 paper 的野心，也並不是發明一個新的 Streaming SQL dialect，而是試圖將 streaming，重新放回標準 SQL 與 relational algebra 的世界裡。只要加入少量 extension，SQL 就可以同時描述：

• bounded batch
• unbounded stream
• changelog
• time-varying relations

這篇論文之所以特別值得閱讀，不只是因為它提出了許多後來深遠影響業界的概念，更因為它其實是一篇建立在多年工業實踐之上的統整性論文。

作者群長期參與 Google Dataflow、Apache Beam、Apache Flink、Apache Calcite 等系統的設計與演化，因此 paper 中許多提案，並非憑空構想，而是來自真實 streaming SQL 系統所遭遇的問題與經驗總結。

有趣的是，並非所有提案最後都被主流系統完整採納。有些概念，例如 Time-Varying Relations、event time、watermark、window table-valued functions，以及 changelog stream，後來幾乎成為整個 streaming SQL 世界的共同語言；而另一些更具野心的提案，則至今仍未成為主流 SQL 標準的一部分。

接下來，我們將依序介紹這篇論文提出的七個核心 proposal。其中，Proposal 1 奠定了整篇論文最重要的理論基礎；而 Proposal 2 至 Proposal 7，則是在此基礎之上，逐步建立完整的 Streaming SQL 語法與語意模型。

在介紹每個 proposal 的同時，我們也會觀察當今主流系統的採納情況，並以 Apache Flink SQL 作為主要範例，看看哪些構想已經成為產業標準，又有哪些構想至今仍停留在論文之中。

Proposal 1: Time-Varying Relations（TVR）

TVR 是這篇論文的核心概念。

在傳統 relational model 中，table 通常被視為某個靜態資料集。query 的輸入是 table，輸出也是 table；而 query result，通常被假設是一個相對穩定、不再變動的結果。

但在 streaming system 裡：

• 資料會持續到達
• state 會持續變化
• query result 也會不斷被修正

於是，paper 提出了一個更根本的觀點：relation 並不是靜態存在，而是一個隨時間持續演化的 semantic object。這個 semantic object，便是 Time-Varying Relation。

而 stream 與 table，則只是 TVR 的不同 representation。

這比先前提到的 stream-table duality 更加激進。它並不是在說 stream 與 table 很像，而是在說 stream 與 table 根本就是同一個東西。

而這個觀點，也徹底改變了 SQL 在 streaming world 中的定位。因為一旦：

• stream = representation of TVR
• table = representation of TVR

那麼 SQL 便不再只是查詢靜態 table 的語言，它開始變成一種描述 TVR 如何隨時間演化的語言。

Proposal 2: Event Time Columns and Watermarks

在此提案，event time 被正式帶入 SQL。

在傳統 SQL 裡，時間通常只是普通欄位；但在 streaming SQL 裡，時間不只是資料的一部分，而是系統判斷「資料是否大致到齊」的依據。paper 因此主張：relation 中應該可以有明確的 event-time column，而這個 column 需要搭配 watermark，讓系統能推理 event-time progress。

paper 對 watermark 的定義很關鍵：watermark 是一個從 processing time 到 event time 的 monotonic function。換句話說，在某個 processing time，系統宣告 event time 已經推進到某個時間點；也就是說，未來應該不會再看到 timestamp 小於或等於該 watermark 的資料。

這也是為什麼 watermark 對 streaming SQL 很重要。沒有 watermark，系統永遠無法知道某個 window 或 GROUP BY 是否可以被視為「足夠完整」；有了 watermark，系統才有機會在 correctness、latency 與 resource cleanup 之間做取捨。

Flink SQL 已經相當完整地支援這個提案。Flink 的 WATERMARK clause 可以在 CREATE TABLE 中宣告 event-time attribute，形式是 WATERMARK FOR rowtime_column AS watermark_strategy_expression；官方文件也說，這個 clause 會把指定 column 標記為 event-time attribute。

例如：

CREATE TABLE orders (
  order_id STRING,
  user_id STRING,
  amount DECIMAL(10, 2),
  -- event-time column
  order_time TIMESTAMP(3),
  -- 表示系統假設資料最多可能晚到 5 秒。
  -- 當目前看到的最大 event time 是 10:00:30 時，watermark 大約會推進到 10:00:25。
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'format' = 'json'
);

這正是 paper 所說的方向：event time 不再是隱藏在 stream metadata 裡的東西，而是 SQL schema 中可以被明確宣告、操作，並與 watermark 一起參與 query semantics 的一等公民。

Proposal 3: Grouping on Event Timestamps

在 batch SQL 裡，GROUP BY 是一件非常自然的事情。因為資料是有限的，系統只需要等所有資料讀完，再做 aggregation 即可。系統知道 input 何時結束，因此也知道 aggregation 何時完成。例如：

SELECT user_id, COUNT(*)
FROM clicks
GROUP BY user_id;

但對 unbounded stream 而言，input 是源源不絕的。這意味著如果沒有額外資訊，GROUP BY 永遠不會完成。

因此論文特別強調，對 unbounded relation 做 grouping 時，必須依賴 event-time column 與 watermark。因為只有 watermark，才能讓系統推理某個 event-time interval 是否已經 sufficiently complete。

例如使用 Flink SQL 對同一個 orders table 進行 event-time grouping：

SELECT
  window_start,
  window_end,
  COUNT(*) AS order_count,
  SUM(amount) AS total_sales
FROM TABLE(
  TUMBLE(
    TABLE orders,
    DESCRIPTOR(order_time),
    INTERVAL '1' MINUTE
  )
)
GROUP BY window_start, window_end;

在這個 query 裡，Flink 會根據 watermark 的推進，決定某個 event-time window 是否可以被視為 sufficiently complete，並觸發 aggregation result。

Proposal 4: Event-Time Windowing Functions

在早期 streaming SQL 系統裡，window 通常被當成特殊版本的 GROUP BY。例如：

GROUP BY TUMBLE(…)

這類語法雖然實用，但 paper 認為 window 被硬塞進 GROUP BY，破壞了 relational semantics。因為在 relational algebra 的世界裡 GROUP BY 本來應該只是對 relation 做 grouping。而 window 其實是在對 relation 進行時間切分（partitioning over time）。

因此，論文提出 windowing 應該被視為 table-valued function（TVF）：接收一個 relation，
並回傳另一個 relation 的 function。這樣一來 windowing 就重新回到 relational model 裡。

Flink SQL 在這部分完成度非常高。目前 Flink 已經完整支援：

• TUMBLE
• HOP
• CUMULATE
• SESSION

等 Windowing TVF。

TUMBLE

SELECT
  window_start,
  window_end,
  COUNT(*) AS order_count
FROM TABLE(
-- TUMBLE(...) 會先產生：帶有 window_start / window_end metadata 的 relation。之後 GROUP BY 才對這個 relation 做 aggregation。
  TUMBLE(
    TABLE orders,
    DESCRIPTOR(order_time),
    INTERVAL '1' MINUTE
  )
)
GROUP BY window_start, window_end;

HOP

SELECT
  window_start,
  window_end,
  SUM(amount) AS total_sales
FROM TABLE(
-- 每 1 分鐘滑動一次，計算最近 5 分鐘的 aggregation。
  HOP(
    TABLE orders,
      DESCRIPTOR(order_time),
      INTERVAL '1' MINUTE,
      INTERVAL '5' MINUTE
  )
)
GROUP BY window_start, window_end;

SESSION

SELECT
  window_start,
  window_end,
  user_id,
  COUNT(*) AS click_count
FROM TABLE(
-- 若 30 分鐘內沒有新 event，則 session 結束。
  SESSION(
    TABLE clicks,
    DESCRIPTOR(click_time),
    INTERVAL '30' MINUTE
  )
)
GROUP BY window_start, window_end, user_id;

CUMULATE

這類 cumulative window，常用於：

• dashboard incremental refresh
• partial aggregation
• progressive reporting

SELECT
  window_start,
  window_end,
  SUM(amount) AS revenue
FROM TABLE(
  CUMULATE(
    TABLE orders,
    DESCRIPTOR(order_time),
    INTERVAL '1' MINUTE,
    INTERVAL '1' HOUR
  )
)
GROUP BY window_start, window_end;

Proposal 5: EMIT STREAM (How)

在 paper 的模型裡，query result 並不一定只是某個時間點的 snapshot table。很多時候，使用者真正想要的是 result 的持續變化過程。例如：

-- 只是 proposal
SELECT
  product_id,
  COUNT(*) AS order_count
FROM orders
GROUP BY product_id
EMIT STREAM;

代表當 aggregation state 發生變化時，持續輸出 refinement。例如：

product_42 → 58
product_42 → 59
product_42 → 61

這也是 paper 對：

• stream-table duality
• TVR
• refinement semantics

的一次具體化。在 Dynamic Tables 的觀點下，這等價於：

• 將 continuously evolving relation，materialize 成 changelog stream。

Proposal 6: EMIT AFTER WATERMARK (When)

在 dashboard, billing 或 reporting 的用例中，使用者並不希望每一次 refinement 都立刻被 downstream 看見。例如：

-- 只是 proposal
SELECT
  window_start,
  window_end,
  COUNT(*) AS order_count
FROM TABLE(
  TUMBLE(
    TABLE orders,
    DESCRIPTOR(order_time),
    INTERVAL '1' MINUTE
  )
)
GROUP BY window_start, window_end
-- 這代表等 watermark 超過 window_end，才 materialize result。
EMIT AFTER WATERMARK;

EMIT AFTER WATERMARK 試圖將 watermark-driven materialization 提升到 SQL 層級，使使用者能夠顯式控制 query result 何時被 materialize，而不是將這個決策隱含在 runtime 或 window operator 中。

雖然 Flink 沒採用這個語法，但其在 event-time window query 中的預設的 trigger 就是 watermark-based emission。

Proposal 7: EMIT AFTER DELAY (When)

如果每次 state change 都 emit， 高流量 query 可能產生大量 updates。對 downstream system 而言，這可能：

• 太頻繁
• 太昂貴
• 太 noisy

因此，我們可以這樣用：

-- 只是 proposal
SELECT
  product_id,
  SUM(amount) AS total_sales
FROM orders
GROUP BY product_id
EMIT STREAM
-- 這代表即使 underlying relation 持續變化，系統也只會每 10 秒 materialize 一次。
AFTER DELAY INTERVAL '10' SECOND;

例如輸出可能是這樣：

10:00:00 → product_42 = 58
10:00:10 → product_42 = 73
10:00:20 → product_42 = 91

paper 更進一步提出：EMIT STREAM, AFTER DELAY, AFTER WATERMARK

可以彼此組合。例如：

• 平時每 10 秒輸出一次 approximate result
• watermark 到達後輸出 on-time result
• late data 再觸發 refinement

這其實已經非常接近 Beam trigger model 與 refinement semantics 的 SQL 化版本。

The Unfinished Proposals of Streaming SQL

回頭來看，《One SQL to Rule Them All》所提出的七個 proposal 中，大部分都已經以不同形式被主流 streaming SQL 系統吸收。然而，仍有三個與 materialization control 相關的 proposal，至今尚未成為主流 SQL 標準的一部分：

• EMIT STREAM
• EMIT AFTER WATERMARK
• EMIT AFTER DELAY

有趣的是，這並不代表它們背後的思想失敗了。事實上，這些 proposal 所試圖解決的問題，正是 Dataflow 模型中的 When 與 How：

• 結果應該在什麼時候被輸出？
• 多次輸出的結果應該如何呈現？

真正困難的地方在於，當這些概念被推進 SQL 之後，它們已經不再只是 query semantics，而開始涉及 execution engine 的設計。例如：

EMIT AFTER DELAY INTERVAL '10' SECOND

這句話究竟代表：

• operator 每 10 秒輸出一次？
• sink 每 10 秒 flush 一次？
• downstream 每 10 秒看到一次更新？
• mini-batch 每 10 秒觸發一次？

不同系統可能有完全不同的解釋，因此很難形成跨引擎的一致標準。

這也是為什麼今天的 Flink 雖然已經實作了大量 Dataflow 的核心思想，但 Flink SQL 對於 When 與 How 的支援，仍然不是完全 declarative 的。

所謂 declarative，指的是使用者只需要描述「想要什麼結果」，而不需要關心系統如何實作。例如：

SELECT …
EMIT AFTER DELAY INTERVAL '10' SECOND

光從 query 本身，我們便能清楚知道：

• 這個查詢的結果，應該每 10 秒被 materialize 一次。

然而在今天的 Flink 中，許多與 When 與 How 相關的行為，並不是直接寫在 SQL 裡，而是分散在 runtime configuration、Trigger、Timer、State 與 sink implementation 等不同層次之中。例如：

SET 'table.exec.mini-batch.allow-latency' = '10 s';

或是在 Java 程式中自行實作 Trigger 與 Timer。

換句話說，系統雖然具備這些能力，但使用者往往需要透過額外設定或程式碼組合，才能表達完整的行為。這種方式更接近 imperative programming：開發者必須告訴系統應該如何執行，而不只是描述想要的語意。

若從 API 的角度來看，Flink 大致可以分成三個層次：

Flink SQL
↓
Table API
↓
DataStream API

最上層的 Flink SQL，主要提供：

• Dynamic Tables
• event time
• watermark
• Window TVF
• changelog semantics

以及各種 SQL 查詢能力。

中間的 Table API 則與 Flink SQL 共用同一套 planner 與 runtime。許多設定，例如：

SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '5 s';

本質上都屬於 Table Layer 的 runtime configuration。

然而，這些設定仍然是在調整系統的執行方式，而不是定義查詢本身的語意。本質上，它們回答的是系統應該如何執行？而不是查詢結果應該如何被 materialize？

真正接近 Dataflow 的 When 與 How 的能力，則位於最底層的 DataStream API。在 DataStream API 中，Flink 提供了這些 execution primitives：

• Trigger
• Timer
• State
• ProcessFunction

開發者可以自行定義：

• watermark 到達時觸發
• processing time 到達時觸發
• count-based trigger
• custom trigger
• timer-based logic

並透過：

• FIRE
• PURGE
• FIRE_AND_PURGE

等 Trigger behavior，組合出各種不同的輸出模式。

因此，discarding mode、accumulating mode，甚至類似 retract/update 的行為，其實都能夠在 DataStream API 中被實作出來。

換句話說，Flink runtime 本身，並不缺乏實現 Dataflow When 與 How 的能力。真正缺少的是將這些能力提升成 SQL 語言的一部分。

因此，今天的 Flink 呈現出一個相當有趣的局面：What（計算什麼）和 Where（在哪個時間範圍計算）已經完整進入 SQL；When（何時輸出）和 How（如何輸出）則仍然有一部分停留在 runtime 與 API 層。

從這個角度來看，《One SQL to Rule Them All》並沒有完全失敗。相反地，它提出的大部分語意模型都已深深影響後來的 Flink、Materialize 與整個 streaming SQL 生態系。真正尚未完成的，只是如何將 Dataflow 的 When 與 How，完整提升為 declarative SQL language 的一部分。

Elaboration for Flink Job

不論你撰寫的是 SQL、Table API，還是 DataStream API，最終都會被編譯成一個或多個 Flink Job，並由 Flink Runtime 負責執行。

因此，當我們討論 Event Time、Watermark、Dynamic Tables、Trigger 或 Changelog 時，它們並不是停留在理論上的抽象概念，而是最終會被轉化為一連串實際運作的 operator、state 與 network shuffle。

現在，讓我們透過一個稍早出現過的 SQL 範例，看看一個 Flink Job 究竟是如何被建構與執行的。首先，我們定義一個 source table：

CREATE TABLE orders (
  order_id STRING,
  user_id STRING,
  amount DECIMAL(10, 2),
  order_time TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'format' = 'json'
);

這裡的 CREATE TABLE orders 並不是在 Flink 內真正建立一張 physical table，而是在 Flink Catalog 中註冊一個對應到 Kafka topic 的 logical table。接著，我們可以撰寫 streaming aggregation：

INSERT INTO minute_sales
SELECT
  window_start,
  window_end,
  COUNT(*) AS order_count,
  SUM(amount) AS total_sales
FROM TABLE(
  TUMBLE(
    TABLE orders,
    DESCRIPTOR(order_time),
    INTERVAL '1' MINUTE
  )
)
GROUP BY window_start, window_end;

這段 SQL 本質上代表持續從 Kafka topic orders 讀取 stream，並不斷計算每分鐘 aggregation。INSERT INTO minute_sales 則表示將這個 continuously evolving result，持續寫入另一個 sink。例如：

CREATE TABLE minute_sales (
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  order_count BIGINT,
  total_sales DECIMAL(10,2)
) WITH (
  'connector' = 'kafka',
  'topic' = 'minute_sales',
  'format' = 'json'
);

因此，整個 topology 實際上是：

Kafka topic: orders
↓
Flink SQL Job
↓
Kafka topic: minute_sales

這裡有一個非常重要的觀念：source 與 sink，通常都必須是 persistent storage。例如：

• Kafka
• Iceberg
• JDBC database
• filesystem

原因是 Flink 本身並不是 database，而更像是一個 continuous query execution engine。而 Flink SQL 中的 CREATE TABLE 很多時候也並不是建立真正的 physical table，而只是在 Catalog 中，
註冊 external system 的 metadata。

另外，一個 query 通常代表一個 Flink job。例如：

INSERT INTO minute_sales
SELECT …

在 production environment 中，通常會被編譯成一個持續執行（continuously running）的 Flink job。只要 Kafka 中還有新的 event 持續進來，這個 job 就會：

• 持續 consume stream
• 持續更新 state
• 持續計算 aggregation
• 持續 output changelog

而這也正是 Dynamic Tables 最重要的思想之一：query result 不再是一次性產生的 static table，而是一個 continuously evolving relational state。

然而，在 query result 持續演化的當下，如果機器故障了怎麼辦？事實上，這正是 Streaming System 早期最困難的問題之一。

2013 年，Google MillWheel 首次系統性地將 persistent state 與 checkpoint 機制引入大型 stream processing system，並藉此實現 exactly-once processing semantics。MillWheel 的許多設計，後來也深深影響了 Dataflow、Beam 與 Flink 等系統。

在 Flink 中，state 會定期被 snapshot 到 checkpoint storage。當機器故障時，Flink 可以從最近一次 checkpoint 恢復 operator state、watermark progress 與 processing progress，並繼續執行 job。例如，在 DataStream API 中，可以透過下列設定開啟 checkpoint：

StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();

env.enableCheckpointing(10000); // every 10 seconds

或是在 Flink configuration 中，以下代表系統每 10 秒建立一次 checkpoint：

execution.checkpointing.interval: 10s

目前常見的 checkpoint storage 例子包括：

execution.checkpointing.storage: filesystem
execution.checkpointing.dir: s3://flink-checkpoints/
execution.checkpointing.dir: hdfs:///flink/checkpoints

在大型 production environment 中，S3 往往是最常見的選擇。因為 checkpoint 本質上就是 state snapshot，而 object storage 具備高耐久性、低成本以及良好的水平擴展能力。

Summary: From MillWheel to Flink

回顧這段歷史，Google 的兩篇經典論文其實分工非常明確。

2013 年的 MillWheel，主要關注的是 execution correctness。它試圖回答的是：

• 系統如何在機器故障後恢復？
• 如何保存 state？
• 如何實現 exactly-once？

因此，MillWheel 花了大量篇幅討論 state management、checkpoint、low watermark、fault tolerance 與 recovery 等工程問題。

而 2015 年的 Dataflow 則選擇站在另一個高度思考問題。它預設底層 execution engine 已經存在，並將注意力集中在時間語意（time semantics）本身。

換句話說，MillWheel 關心的是「如何保存正確的 state？」而 Dataflow 關心的則是：「什麼才是正確的結果？」從今天回頭看，這兩個問題其實密不可分。Trigger、Accumulation 與 Refinement 都依賴 state 的持久化；而 state 的恢復方式，又會直接影響結果的正確性。

於是，一個自然的問題浮現：「如果能夠同時擁有 MillWheel 的 state 與 fault tolerance，再加上 Dataflow 的時間語意，會得到什麼樣的系統？」

Apache Flink 可以說是這個問題最具代表性的答案之一。它不僅吸收了 MillWheel 在 state、checkpoint 與 exactly-once 上的工程經驗，也採納了 Dataflow 的 Event Time、Watermark、Trigger 與 Accumulation 等時間語意模型，將 execution correctness 與 semantic correctness 統一到同一個系統之中。

而本章開頭曾經提到：在 2013 到 2019 這段歷史中，發生了三次極其重要的抽象融合。

第一個融合，是 Google Dataflow 的 Unifying Batch and Streaming。

在此之前，batch 與 streaming 長期被視為兩種截然不同的計算模式。然而 Dataflow 透過 What、Where、When 與 How 這套統一的模型，重新定義了兩者之間的關係。Batch 不再是一套獨立的系統，而可以被理解為 streaming 的特殊情況；streaming 則成為更一般化的資料處理模型。

第二個融合，是 Flink Dynamic Tables 的 Stream-Table Duality。

在此之前，stream 屬於事件與消息的世界；table 則屬於資料庫與 SQL 的世界。兩者雖然經常共存，卻很少被放進同一套理論框架之中。

Dynamic Tables 第一次讓資料庫世界與 streaming 世界開始共用同一套語意：stream 是 changelog；table 是 materialized state。兩者不再是完全不同的資料型態，而是同一個現象的兩種觀察方式。

而第三個融合，至今仍在持續發展之中。

《One SQL to Rule Them All》與 TVR（Time-Varying Relations）更進一步提出了一個比 Stream-Table Duality 更激進的觀點。在 TVR 的世界裡，stream 與 table 是同一個 semantic object 在不同時間尺度下的呈現方式。

換句話說，它不再主張：

• stream 與 table 很像；

而是主張：

• stream 與 table 本質上就是同一件事。

這正是第三次抽象融合：Unifying Stream, Table and SQL

從 MillWheel 到 Dataflow，從 Dynamic Tables 到 TVR，下層有新系統的誕生，上層也有一連串抽象層次的提升。

一旦 stream 與 table 成了 TVR，這趟乘載抽象與想像的 streaming 之旅又往前推進了一大步。stream、batch、table 與 SQL 這些原本彼此分離的概念，也開始逐漸被統一進同一套 worldview 之中。

(To be continued …)

Appendix

1. 極客 29 | Dataflow（一）：正确性、容错和时间窗口
2. 極客 30 | Dataflow（二）：MillWheel，一个早期实现
3. MillWheel: Fault-Tolerant Stream Processing at Internet Scale: https://static.googleusercontent.com/media/research.google.com/en//pubs/archive/41378.pdf
4. The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing: https://static.googleusercontent.com/media/research.google.com/en//pubs/archive/43864.pdf
5. Apache Beam: https://beam.apache.org/about/
6. Dynamic Tables: https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/concepts/dynamic_tables/
7. One SQL to Rule Them All: An Efficient and Syntactically Idiomatic Approach to Management of Streams and Tables: https://arxiv.org/pdf/1905.12133
8. One SQL to Rule Them All — Fabian Hueske: https://www.youtube.com/watch?v=uJtqGkIxGhc
9. Apache Flink™ vs Apache Kafka™ Streams vs Apache Spark™ Structured Streaming — Comparing Stream Processing Engines: https://www.onehouse.ai/blog/apache-spark-structured-streaming-vs-apache-flink-vs-apache-kafka-streams-comparing-stream-processing-engines
10. Arroyo: https://www.arroyo.dev/blog/why-not-flink/