串流資料管線 — 面試深入解析

範圍 — 資料工程師面試會問到的串流處理全貌：何時該用串流、參考架構、時間語意與 watermark、視窗與遲到資料、狀態與 checkpoint、端到端 exactly-once、串流 join、背壓與擴展、各引擎（Flink、Spark Structured Streaming、Kafka Streams、Beam）、即時服務與維運。Kafka broker 內部機制不在這裡 —— 它們屬於 kafka.md。 另見：../faq_DE_interview.md —— 總覽與考前檢查清單；./kafka.md —— 大多數串流底下的那份 log；./batch_pipeline.md —— 同一條管線的批次那一側；./spark.md —— Spark 內部機制；./flink.md —— Flink 的 runtime、API 與設定的深入內容；../flink/faq_flink.md 與 ../stream/faq_stream.md —— 較早的筆記。

每一節都先給你會說出口的那句話，再講機制，然後是面試官下一步會追問的取捨。 程式碼以 Flink 1.19+ / 2.x、Spark 3.5+ 與 Kafka Streams 3.x 為準；版本有差的地方會特別註明。

1) 批次 vs 串流 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

1-1) 什麼時候值得用串流

只有當有人會根據結果採取行動、而且需要比批次更快拿到時，才用串流 —— 串流買到的是 延遲，付出的代價是永遠在線的基礎設施與更難的正確性。

好的串流理由：詐欺與濫用偵測、告警與監控、即時營運 dashboard、個人化特徵、 系統之間的 CDC 複製。 薄弱的理由：「日報應該要即時」，但根本沒人一天看超過一次 —— 每小時一次的批次 更便宜也更簡單。

面試訊號：選擇之前先問「誰會讀這個結果，它可以多舊？」

1-2) Micro-batch vs 逐筆處理

Micro-batch 引擎把一小批紀錄收集起來，對它跑一個迷你批次 job；逐筆（record-at-a-time） 引擎則在每筆紀錄到達時就把它推過各個運算子。

Spark 也有 continuous processing 模式，但它仍標示為實驗性，而且只支援類似 map 的運算 —— 不要把它當成低延遲的答案。

1-3) Lambda vs Kappa 架構

Lambda 用一個批次層負責準確、一個速度層負責新鮮度，再把兩者合併；Kappa 只跑串流， 重新處理時就重播 log。

Lambda                                      Kappa
                ┌─► batch layer ─┐                     ┌─► stream job v1 ─► serving
source ─► log ──┤                ├─► serve   source ─► log
                └─► speed layer ─┘                     └─► stream job v2 (replay) ─► serving

現代做法偏向 Kappa：一個能用同一份邏輯跑批次與串流的引擎（Flink、Spark、Beam）， 原始串流封存在物件儲存或 lakehouse 表裡，批次回補用同一份程式碼讀那份封存。

2) 參考架構 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

2-1) 標準的管線

來源把事件寫進一份持久的 log，有狀態的處理器把它們轉成結果，sink 再把結果提供給 dashboard、服務或 lakehouse。

 Apps / devices ─► producers ─┐
                              ├─► Kafka (raw topics) ─► Stream processor ─┬─► Kafka (derived topics)
 OLTP DB ─► CDC (Debezium) ───┘         │               (Flink / Spark /   ├─► Real-time OLAP (Pinot, Druid, ClickHouse)
                                        │                Kafka Streams)    ├─► Serving store (Redis, Cassandra, DynamoDB)
                                        │               state + checkpoints└─► Lakehouse table (Iceberg / Delta / Hudi)
                                        └─► archive to object storage (replay, batch backfill)

各個元件以及它們存在的理由：

• 持久 log（Kafka、Kinesis、Pub/Sub）—— 讓 producer 與 consumer 解耦，並讓來源 可以重播，每一個復原與重新處理的方案都依賴這一點。
• Schema Registry —— producer 與 consumer 對事件的形狀達成一致；見 schema_design.md。
• 串流處理器 —— 過濾、補充（enrichment）、彙總、join；持有狀態。
• 衍生 topic —— 其他團隊消費的結果，所以一個 job 的輸出就是另一個 job 的輸入。
• Sink —— 依存取模式選擇：依 key 查找 → key-value 儲存，任意切片的彙總 → 即時 OLAP，歷史與分析 → lakehouse 表。

2-2) 以 CDC 作為來源

變更資料擷取（CDC）讀取資料庫自己的變更 log（MySQL binlog、PostgreSQL WAL），把每一次 insert、update、delete 轉成事件，而不必查詢資料表。

• 常見答案是跑在 Kafka Connect 上的 Debezium；每張表變成一個以主鍵為 key 的 topic， 帶有 before / after 映像與一個 op 欄位（c、u、d，快照則是 r）。
• 它先做一次初始快照，再從它記錄下的 log 位置開始串流變更。
• 刪除會先送出一個 delete 事件，再跟著一個 tombstone（值為 null），讓 compacted topic 可以丟掉那個 key。
• Outbox 模式 —— 服務在同一個本地交易中寫入業務資料列和一筆 outbox 資料列； CDC 負責發佈 outbox。它避免了雙寫問題（DB 提交成功但 Kafka 發佈失敗，或反過來）。

2-3) 選擇 sink

Sink 決定你能提供的投遞保證（第 6 節）：依 key upsert 的 sink 能讓 at-least-once 看起來像 exactly-once；只能 append 的 sink 則不行。

3) 時間語意與 Watermark ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

3-1) 事件時間、擷取時間、處理時間

事件時間是事情發生的時間；處理時間是引擎看到它的時間。正確的彙總要用事件時間， 因為處理時間每次重播都會變。

事件時間與處理時間之間的差距（skew）從來不是常數：行動裝置會離線、某個 partition 會落後、producer 會重試。這正是 watermark 存在的全部理由。

3-2) Watermark

Watermark 是串流中的一個標記，意思是「不會再有時間戳 ≤ T 的事件了」；當 watermark 越過事件時間視窗與計時器時，視窗才會關閉、計時器才會觸發。

常見策略是 bounded out-of-orderness（有界亂序）：watermark = 目前看到的最大事件時間 − 一個固定的上限。每個平行的來源實例各自送出 watermark，而有多個輸入的運算子取它們的 最小值。

// java — Flink 1.19+: watermarks for a Kafka source
WatermarkStrategy<Ride> watermarks = WatermarkStrategy
        .<Ride>forBoundedOutOfOrderness(Duration.ofSeconds(10))   // tolerate 10 s of disorder
        .withTimestampAssigner((ride, recordTs) -> ride.getEventTimeMillis())
        .withIdleness(Duration.ofMinutes(1));                     // see 3-3

DataStream<Ride> rides = env.fromSource(kafkaSource, watermarks, "rides");

Watermark 是一種啟發式估計，不是保證：它用完整性換延遲，而落在它之後的一切 都需要一個遲到資料策略（第 4-3 節）。

3-3) 閒置來源與 watermark 對齊

因為運算子取輸入 watermark 的最小值，一個停止送資料的 partition 就會凍結整個 job 的 watermark —— 所有視窗都不再關閉。

• Idleness（閒置） —— withIdleness(...) 在安靜一段時間後把該 split 標成閒置， 讓它在資料再次到達前不參與最小值計算。
• Watermark alignment（對齊）（Flink 1.15+，withWatermarkAlignment）—— 處理相反的 問題：某個來源遠遠領先其他來源，讓下游運算子必須把它的資料緩衝在狀態裡。對齊會暫停 那些領先太多的來源。
• 一個 partition 數多於活躍 key 數的 topic，或一個讀某個 partition 比另一個快的回補， 都會同時觸發這兩個問題；在 Flink UI 檢查每個 subtask 的 watermark。

4) 視窗、Trigger 與遲到資料 ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

4-1) 視窗類型

視窗把無限的串流切成可以彙總的有限區塊；類型決定這些區塊是固定的、重疊的， 還是由活動的間隔來定義。

// java — Flink 1.19+: one-minute tumbling event-time windows per city
rides.keyBy(Ride::getCity)
     .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
     .aggregate(new CountAndSumFare())          // incremental: keeps one accumulator per window
     .sinkTo(resultSink);

優先用增量的 aggregate / reduce，而不是會緩衝每個事件的 process 視窗函式： 每個視窗的狀態就從「所有事件」降為「一個累加器」。

4-2) Trigger：視窗何時輸出

Trigger 決定視窗的結果何時輸出；預設的事件時間 trigger 只觸發一次，就在 watermark 越過視窗結束時間的時候。

• 提早觸發（early firing） —— 在視窗關閉前輸出推測性的結果（每 10 秒處理時間一次）， 讓 dashboard 不會在一小時長的視窗裡整段空白。
• 延後觸發（late firing） —— 在允許的遲到時間內，每個遲到事件都輸出一次更新後的結果。
• 第一次之後的每次觸發都是同一個視窗的修正（refinement），所以 sink 必須能接受它： 依 (key, window_start) 覆寫，或消費 retraction 串流。
• Beam 把這幾個問題講得很明白：計算什麼（轉換）、在事件時間的哪裡（視窗）、 在處理時間的何時（trigger），以及修正之間如何關聯（accumulating vs discarding panes）。

4-3) 遲到資料策略

遲到事件是指在 watermark 已經越過其視窗之後才到達的事件；你可以丟掉它、把它導到 旁邊，或更新已經輸出的結果。

// java — Flink: keep windows 5 extra minutes, send anything later to a side output
final OutputTag<Ride> lateRides = new OutputTag<Ride>("late-rides") {};

SingleOutputStreamOperator<CityCount> counts = rides
        .keyBy(Ride::getCity)
        .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
        .allowedLateness(Duration.ofMinutes(5))
        .sideOutputLateData(lateRides)
        .aggregate(new CountAndSumFare());

DataStream<Ride> tooLate = counts.getSideOutput(lateRides);   // e.g. write to a reconciliation table

5) 狀態 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

5-1) Keyed state vs operator state

Keyed state 依串流的 key 分區，只有帶著那個 key 的紀錄看得到；operator state 屬於 一個平行實例 —— 來源已讀到的 Kafka offset 就是經典例子。

// java — Flink keyed state: alert when a card's running spend exceeds a limit
public class SpendTracker extends KeyedProcessFunction<String, Payment, Alert> {
    private transient ValueState<Double> total;

    @Override
    public void open(OpenContext ctx) {              // Flink 1.19+; older versions: open(Configuration)
        StateTtlConfig ttl = StateTtlConfig.newBuilder(Duration.ofHours(24)).build();
        ValueStateDescriptor<Double> desc = new ValueStateDescriptor<>("total", Double.class);
        desc.enableTimeToLive(ttl);                  // bound state: idle cards expire
        total = getRuntimeContext().getState(desc);
    }

    @Override
    public void processElement(Payment p, Context ctx, Collector<Alert> out) throws Exception {
        double current = total.value() == null ? 0.0 : total.value();
        current += p.getAmount();
        total.update(current);
        if (current > 10_000.0) {
            out.collect(new Alert(p.getCardId(), current));
        }
    }
}

5-2) State backend

State backend 決定即時狀態放在哪裡、怎麼做快照：放在 JVM heap 求快，或放在本機磁碟上 的嵌入式 RocksDB 以容納比記憶體更大的狀態。

不論即時狀態放在哪個 backend，checkpoint 本身都寫到持久儲存（S3、HDFS、GCS）。

5-3) 狀態 TTL 與狀態大小

除非有東西把它移除，串流狀態會無限成長：視窗關閉、watermark 越過，或明確的 TTL。

• 無界的 key —— 沒有視窗的 GROUP BY user_id 會為每個出現過的使用者保留一筆。 設定 TTL（StateTtlConfig、Flink SQL table.exec.state.ttl）或把彙總加上視窗。
• 沒有時間上限的串流-串流 join 會永遠保留兩邊（第 7-1 節）。
• 去重集合需要一個時間範圍：「丟掉 24 小時內看過的重複」，而不是「永遠」。
• 狀態成長的徵兆：checkpoint 大小與時間持續上升、RocksDB 磁碟逐漸塞滿、 每週復原都變得更慢。
• TTL 是正確性的決定，不只是維運的決定：過期的 key 表現得像新的一樣，所以超過 TTL 後回來的使用者會被當成新使用者。

6) 容錯與 Exactly-Once ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

6-1) Checkpoint：分散式快照

Checkpoint 是每個運算子狀態加上其對應來源 offset 的一致性快照；故障時，job 還原最後 一個 checkpoint，並從那些 offset 重播來源。

Flink 實作的是 Chandy-Lamport 演算法的一種變體（asynchronous barrier snapshotting， 非同步屏障快照）：

 source ──[e5][e4][ B n ][e3][e2]──► operator ──► sink

1. The checkpoint coordinator injects barrier n into every source; each source records its offsets.
2. Barriers flow with the data. An operator that has received barrier n on all inputs snapshots
   its state (asynchronously) and forwards the barrier.
3. When every sink has acknowledged barrier n, checkpoint n is complete.
4. Failure → restart all tasks from checkpoint n: restore state, rewind sources to n's offsets.

Barrier n 之前的一切都在快照裡；之後的一切都會被重播。這就是狀態一致的原因：沒有任何 事件在狀態中被算兩次，也沒有任何事件從狀態中遺失。

// java — Flink: checkpoint every minute, exactly-once state semantics
env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);
env.getCheckpointConfig().setCheckpointTimeout(10 * 60_000);

6-2) 對齊 vs 非對齊 checkpoint

使用對齊（aligned）checkpoint 時，有多個輸入的運算子要等每個輸入都收到 barrier 才做 快照；在背壓下這段等待可能讓 checkpoint 卡很久，非對齊（unaligned）checkpoint 就是為了 避免這點。

At-least-once checkpoint 完全跳過對齊：在其輸入的 barrier 之後才到的紀錄可能被處理、 然後又被重播，所以可能被算兩次。

6-3) Checkpoint vs savepoint

Checkpoint 是自動的，用來從故障中復原；savepoint 由你觸發，用來在不遺失狀態的情況下 停止、升級、擴縮或遷移 job。

給每個有狀態的運算子一個穩定的 .uid("...")：savepoint 的狀態是依 id 對應到運算子的， 而自動產生的 id 在 job 圖改變時就會變。

6-4) 端到端 exactly-once ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

串流引擎裡的「exactly-once」指的是每個事件只影響狀態一次；要在輸出也做到，需要 三樣東西：可重播的來源、一致的 checkpoint，以及支援交易或冪等的 sink。

• 交易式 sink —— 輸出先暫存，只有在 checkpoint 提交時才變得可見。下游讀取者必須 只讀已提交的資料（Kafka isolation.level=read_committed）。 代價：輸出每個 checkpoint 才可見一次，所以延遲 ≥ checkpoint 間隔。
• 冪等 sink —— 重播時以相同的值寫入相同的 key（依 (key, window_start) upsert、 確定性的 id）。延遲較低，但讀取者可能短暫看到一個之後會被以相同值重寫的結果。
• Sink 以外的副作用（寄信、呼叫 API）永遠不在保證範圍內 —— 用去重 key 讓它們冪等。
• 沒有交易式或冪等的 sink 時，誠實的描述是 at-least-once。

6-5) 兩階段提交 sink

兩階段提交 sink 每個 checkpoint 開一個交易，barrier 到達時預提交（pre-commit）， 只有在整個 checkpoint 成功後才正式提交。

barrier n arrives at sink ──► pre-commit txn n (flush, keep it open) ──► start txn n+1
checkpoint n completes     ──► commit txn n          (now visible to read_committed readers)
failure before completion  ──► abort txn n, restore checkpoint n-1, replay

// java — Flink KafkaSink with exactly-once delivery
KafkaSink<String> sink = KafkaSink.<String>builder()
        .setBootstrapServers("broker:9092")
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                .setTopic("ride-counts")
                .setValueSerializationSchema(new SimpleStringSchema())
                .build())
        .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
        .setTransactionalIdPrefix("ride-counts-job")      // must be unique per job
        .setProperty("transaction.timeout.ms", "900000")  // ≤ broker transaction.max.timeout.ms
        .build();

陷阱：producer 的 transaction.timeout.ms 必須涵蓋最長的 checkpoint 加上重啟時間，否則 broker 會中止一個已預提交的交易，其資料就遺失了；而且它不能超過 broker 的 transaction.max.timeout.ms（預設 15 分鐘）。

7) 串流 Join 與資料補充 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

7-1) 串流-串流 join

Join 兩個無界串流需要一個時間上限 ——「付款在訂單後 15 分鐘內到達」—— 引擎才知道 什麼時候可以忘掉它正在緩衝的資料列。

-- Flink SQL: match each order to its payment within 15 minutes (interval join)
SELECT o.order_id, o.amount, p.payment_id, p.pay_time
FROM   orders o, payments p
WHERE  o.order_id = p.order_id
  AND  p.pay_time BETWEEN o.order_time AND o.order_time + INTERVAL '15' MINUTE;

Outer join 多了一個問題：什麼時候輸出「沒有付款的訂單」？只有在 watermark 越過區間結束 之後，所以 outer join 中未匹配的資料列會延遲「區間長度 + watermark 上限」。基於同樣的理由， Spark 的 outer 串流-串流 join 要求兩邊都有 watermark 並帶時間條件。

7-2) 串流-表格資料補充

資料補充（enrichment）是為每個事件加上維度屬性（使用者等級、商家類別）；你可以在外部 儲存查找，或把那張表當成 job 內的狀態保存。

Lookup join 看到的是處理時間當下的表，所以重播時會用今天的屬性去補充舊事件； temporal join（7-3）解決這個問題。

7-3) Temporal join

Temporal join 把每個事件 join 到在事件時間當下有效的那個表格版本 —— 等同於串流版的 SCD Type 2 查找。

-- Flink SQL: price each order at the exchange rate valid when it was placed
-- currency_rates must be a versioned table: a PRIMARY KEY and an event-time attribute with a watermark
SELECT o.order_id,
       o.amount * r.rate AS amount_usd
FROM   orders AS o
JOIN   currency_rates FOR SYSTEM_TIME AS OF o.order_time AS r
  ON   o.currency = r.currency;

重播會得到相同的答案，因為查找是以事件時間為準，而不是 join 執行的那一刻。

7-4) Broadcast state

Broadcast state 把一個小而變化緩慢的串流 —— 規則、門檻、功能開關 —— 送到每個平行實例， 讓每一個都能把它套用到自己那一份主串流上。

• 典型用法：詐欺 job 的規則在執行期更新，不需要重新部署。
• 每個實例都持有完整一份，所以只適合小資料。
• Broadcast 那一側在每個實例上必須以完全相同的方式處理；每個 key 的邏輯留在 keyed 那一側。

7-5) 去重

Producer 會重試、來源會重播，所以重複資料是常態；在一個有界的時間範圍內，依唯一的 事件 id 去重。

-- Flink SQL: keep the first event per event_id (the planner recognises this pattern)
SELECT event_id, user_id, event_time
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY event_id ORDER BY event_time ASC) AS rn
  FROM   clicks
)
WHERE rn = 1;

# python — Spark 3.5+: drop duplicates seen within the watermark delay
deduped = (events.withWatermark("event_time", "1 hour")
                 .dropDuplicatesWithinWatermark(["event_id"]))

• id 必須在 producer 端指定 —— 擷取之後才產生的 id 每次重播都不一樣。
• 限制狀態大小：狀態 TTL（Flink SQL）或 watermark（Spark）決定 id 被記住多久。 在那個範圍之後才到的重複資料會直接通過。

8) 背壓、擴展與熱點 Key ★★★☆☆優先度 3／5 — 值得會 — 多半是必備模式的變形

8-1) 背壓

背壓（backpressure）是一個慢的運算子或 sink 讓它上游的一切都跟著變慢，而不是一路緩衝 到記憶體耗盡。

• Flink —— 有界的網路緩衝區：當下游 task 的緩衝區滿了，上游 task 就送不出去而變慢， 一路回推到來源。Web UI 會顯示每個 task 的背壓與忙碌時間。
• Spark —— 每個 micro-batch 只取有限的量：maxOffsetsPerTrigger（Kafka）或 maxFilesPerTrigger（檔案）。
• Kafka Streams —— consumer 單純地 poll 得比較慢。
• 不論哪一種，積壓最後都會變成 Kafka 裡的 consumer lag，而那正是它該在的地方： 由 log 吸收突發流量。

怎麼診斷：找出第一個忙碌、而它下一個卻不忙的運算子 —— 那就是瓶頸。常見原因：慢的 外部查找（用 async I/O）、傾斜的 key、partition 太少的 sink，或 RocksDB 裡狀態很大但 記憶體不夠。

8-2) 擴展與重新擴縮

平行度的上限是來源的 partition 數；keyed state 被切成 key group，重新擴縮時就是把整個 key group 重新分配給新的一組實例。

• Kafka 來源每個 partition 最多一個活躍讀取者，所以平行度超過 partition 數時，多出來的 來源實例會閒置（下游運算子仍可以更寬）。
• Flink 把每個 key 雜湊到 maxParallelism 個 key group 之一；每個 subtask 擁有一段 連續的範圍。重新擴縮 = 做一個 savepoint、以新的平行度重啟，每個 subtask 載入它的新範圍。
• maxParallelism 固定了 key group 的數量，除非改寫狀態（State Processor API），否則 一個 job 的 keyed state 無法改變它。在第一次正式部署之前就刻意設定好。
• 自動擴縮：Flink Kubernetes Operator 的 autoscaler、Dataflow 的自動擴縮，Kafka Streams 則是增加實例（partition 會被重新分配）。

8-3) 熱點 key

一個承載大部分流量的 key 會讓一個 subtask 卡在 100%，其他的卻閒著，再多的平行度都沒用。

9) 各引擎 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

9-1) 比較

9-2) Flink：DataStream 與 Table / SQL

Flink 在同一個 runtime 上有兩套主要 API：完全掌控狀態與計時器的 DataStream API，以及 由 planner 最佳化的宣告式 Table API / SQL。

• DataStream —— keyBy、視窗、帶狀態與事件時間計時器的 KeyedProcessFunction、 side output、async I/O。用在自訂邏輯（詐欺規則、複雜事件處理）。
• Table API / SQL —— 動態表：串流就是一張不斷變化的表，對它的查詢會產生一份 changelog （+I 插入、-U / +U 更新前/後、-D 刪除）。
• 只會 append 的結果（視窗彙總、過濾）可以寫到任何 sink；會更新的結果（沒有視窗的 GROUP BY）需要帶主鍵的 upsert sink（upsert-kafka、帶 key 的 JDBC、Paimon / Hudi）。
• CEP 函式庫 —— 在事件序列上做模式比對（SQL 裡的 MATCH_RECOGNIZE）。

9-3) Spark Structured Streaming

Structured Streaming 把串流當成一張無界的表，對每個 trigger 一個 micro-batch，增量地 執行你的批次 DataFrame 查詢。

# python — Spark 3.5+: one-minute ride counts per city, written to Delta
from pyspark.sql import functions as F

rides = (spark.readStream.format("kafka")
              .option("kafka.bootstrap.servers", "broker:9092")
              .option("subscribe", "rides")
              .option("maxOffsetsPerTrigger", 500000)              # bound each micro-batch
              .load()
              .select(F.from_json(F.col("value").cast("string"), ride_schema).alias("r"))
              .select("r.*"))

counts = (rides.withWatermark("event_time", "10 minutes")
               .groupBy(F.window("event_time", "1 minute"), "city")
               .agg(F.count("*").alias("rides"), F.sum("fare").alias("revenue")))

# append: each window is written once, after the watermark passes its end
(counts.writeStream.outputMode("append")
       .format("delta")
       .option("checkpointLocation", "s3://bucket/chk/rides_per_city")
       .trigger(processingTime="1 minute")
       .start("s3://bucket/tables/rides_per_city"))

# python — updating results into Delta: foreachBatch + MERGE, idempotent per batch
from delta.tables import DeltaTable

def upsert(batch_df, batch_id):
    target = DeltaTable.forPath(spark, "s3://bucket/tables/rides_per_city_live")
    (target.alias("t")
           .merge(batch_df.alias("s"),
                  "t.city = s.city AND t.window = s.window")
           .whenMatchedUpdateAll()
           .whenNotMatchedInsertAll()
           .execute())

(counts.writeStream.outputMode("update")
       .foreachBatch(upsert)
       .option("checkpointLocation", "s3://bucket/chk/rides_per_city_live")
       .start())

• Checkpoint 位置存放 offset、commit log 與狀態；刪掉它，或以不相容的方式改變查詢 （加入有狀態運算子、改變分組 key），就代表要從頭開始。
• trigger(availableNow=True)（3.3+）處理目前所有可用的資料後就停止 —— 把串流程式碼 當成增量批次 job 來跑。

9-4) Kafka Streams：KStream 與 KTable

Kafka Streams 是函式庫，不是叢集：你的應用程式從 Kafka 讀取，把狀態放在以 changelog topic 為後盾的本地 RocksDB，再寫回 Kafka。

// java — Kafka Streams 3.x: rides per city per minute, 30 s grace for late events
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Ride> rides = builder.stream("rides", Consumed.with(Serdes.String(), rideSerde));

KTable<Windowed<String>, Long> perCity = rides
        .groupBy((key, ride) -> ride.getCity(), Grouped.with(Serdes.String(), rideSerde))
        .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(1), Duration.ofSeconds(30)))
        .count();

perCity.toStream()
       .map((windowedCity, count) -> KeyValue.pair(
               windowedCity.key() + "@" + windowedCity.window().start(), count))
       .to("rides-per-city", Produced.with(Serdes.String(), Serdes.Long()));

• Join 需要 co-partitioning：兩個 topic 有相同的 key 與相同的 partition 數；重新指定 key（groupBy、selectKey）會建立一個內部的 repartition topic。
• processing.guarantee=exactly_once_v2 用 Kafka transaction 做 read-process-write。
• 多啟動幾個應用程式實例就能擴展；partition 會在它們之間重新平衡，被搬走的 task 從 changelog 還原狀態（standby replica 能縮短這段時間）。

9-5) Apache Beam 與 Dataflow

Beam 是一套程式設計模型 —— PCollection、視窗、trigger、watermark —— 跑在你選的 runner 上；大家說的通常是託管 runner Google Cloud Dataflow。

• 批次與串流共用一份 pipeline 定義；由 runner（Dataflow、Flink、Spark）執行。
• 它的模型把 what / where / when / how 這幾個問題講得最清楚（第 4-2 節）。
• Dataflow 加上自動擴縮、動態工作重新平衡，以及託管的 shuffle / streaming engine。
• 取捨：可移植性與託管服務，換來較少的調整旋鈕與各 runner 特有的缺口。

10) 串流 SQL ★★★☆☆優先度 3／5 — 值得會 — 多半是必備模式的變形

10-1) Flink SQL 的視窗彙總

串流 SQL 先宣告帶有 watermark 的來源，再跑一個看起來很普通的查詢，其視窗來自 table-valued function。

-- Flink SQL 1.19+
CREATE TABLE rides (
  ride_id     STRING,
  city        STRING,
  fare        DECIMAL(10, 2),
  event_time  TIMESTAMP(3),
  WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'rides',
  'properties.bootstrap.servers' = 'broker:9092',
  'properties.group.id' = 'rides-per-city',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

CREATE TABLE rides_per_city (
  window_start  TIMESTAMP(3),
  window_end    TIMESTAMP(3),
  city          STRING,
  rides         BIGINT,
  revenue       DECIMAL(38, 2)
) WITH (
  'connector' = 'kafka',
  'topic' = 'rides-per-city',
  'properties.bootstrap.servers' = 'broker:9092',
  'format' = 'json'
);

INSERT INTO rides_per_city
SELECT window_start, window_end, city, COUNT(*) AS rides, SUM(fare) AS revenue
FROM   TABLE(TUMBLE(TABLE rides, DESCRIPTOR(event_time), INTERVAL '1' MINUTES))
GROUP  BY window_start, window_end, city;

• TUMBLE、HOP、CUMULATE 與 SESSION（1.19+ 起為 TVF）就是視窗函式。
• 視窗彙總是只會 append 的，所以一般的 kafka sink 就行；同一個查詢若沒有視窗， 就會產生更新，需要帶主鍵的 upsert-kafka。

11) 提供即時結果 ★★★☆☆優先度 3／5 — 值得會 — 多半是必備模式的變形

11-1) 即時 OLAP 儲存

即時 OLAP 儲存直接從 Kafka 擷取，並在次秒級時間內回答彙總查詢，這正是建立在原始事件上 的即時 dashboard 所需要的。

查詢模式固定時（一個 dashboard 方塊）在串流 job 裡先預彙總；使用者會依任意維度切片時， 把原始事件擷取進 OLAP 儲存。

11-2) 串流寫入 lakehouse 表

串流寫進 Iceberg、Delta 或 Hudi，讓串流資料有一個持久、可查詢、具 ACID 提交的家 —— 代價是分鐘級的新鮮度與小檔案問題。

• 每次提交（一個 Flink checkpoint、一個 Spark micro-batch）都會新增檔案；提交越頻繁小檔案 越多，所以要排程壓實（compaction）（Delta OPTIMIZE、Iceberg rewrite_data_files） 與 snapshot 過期清理。
• Upsert / CDC —— Hudi merge-on-read、Iceberg v2 列級刪除、Delta MERGE，以及 Apache Paimon（為 Flink 串流 upsert 而生）依 key 套用變更。
• 提交間隔決定新鮮度：一分鐘一次的 checkpoint 代表資料大約在到達一分鐘後可以查詢。
• 這讓 Kappa 式的系統不需要另一份批次副本就能回答歷史問題。

12) 維運 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

12-1) 要監控什麼

12-2) 故障情境

12-3) 部署新版本與重新處理

• 相容的變更 —— 用 savepoint 停止、部署、從 savepoint 還原。穩定的運算子 uid 讓這件事 可行。
• 不相容的變更（新的彙總 key、新的狀態 schema）—— 讓新 job 和舊 job 並行，從較早的 offset 寫到一個新的輸出表 / topic，比對之後切換讀取者並淘汰舊 job。這就是 Kappa 的 重新處理手法。
• 回補 —— retention 涵蓋時從 Kafka 重播；否則把同一份邏輯當批次 job 跑在封存的原始資料上 （Flink 與 Spark 都能用批次模式跑這份程式碼）。

13) 實戰設計：每城市即時行程數 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

題目：「做一個 dashboard，顯示每個城市開始的行程數與營收，一分鐘內更新， 涵蓋過去 24 小時。」

先釐清：事件量、新鮮度 SLA、數字是否可以因遲到事件而修正、使用者會往回看多遠、 以及是否會依其他維度過濾。

 rider / driver apps ─► API ─► Kafka "ride_events" (key = ride_id, Avro + Schema Registry)
                                     │
                                     ▼
                     Flink job (event time, 30 s bounded out-of-orderness)
                       1. dedupe by event_id (24 h state TTL)
                       2. enrich with city from zone (broadcast state, small table)
                       3. 1-minute tumbling window per city: count, sum(fare)
                       4. allowed lateness 10 min → updated results keyed by (city, window_start)
                                     │
                ┌────────────────────┼──────────────────────────┐
                ▼                    ▼                          ▼
   upsert-kafka "city_minute"   Pinot table (upsert on      Iceberg table (raw events,
   (other consumers)             city + window_start)        hourly compaction) → batch
                                     │                        reconciliation and history
                                     ▼
                               dashboard (queries Pinot)

要說出口的決策：

1. 事件時間搭配 watermark，因為 app 事件會因為連線差而遲到。
2. 結果以 (city, window_start) 為 key 做 upsert，讓遲到事件修正某一分鐘，而不是產生 重複的資料列；dashboard 永遠讀到最新值。
3. 寫到 Pinot 用 at-least-once + 冪等 upsert，而不是付出 exactly-once Kafka 輸出的交易延遲； 重播造成的重複會覆寫同一個 key。
4. 依 producer 指定的 event_id 去重，因為行動客戶端會重試。
5. 原始事件寫進 Iceberg，作為歷史資料，並用每晚的批次對帳修正那些超過允許遲到時間才到的資料。
6. 擴展：Kafka partition 依尖峰吞吐量規劃；視窗以城市為 key，如果某個城市佔大宗，就改用 (city, zone) 為 key，在查詢時再彙總。
7. 維運：對 consumer lag、watermark 延遲與 checkpoint 失敗告警；格式錯誤的事件送到 dead-letter。

14) 快問快答 ★★★☆☆優先度 3／5 — 值得會 — 多半是必備模式的變形

Q：Kafka 是串流處理器嗎？ Kafka 是 log；Kafka Streams 與 ksqlDB 才是跑在它上面的處理器。Broker 本身只負責儲存與 提供紀錄。

Q：為什麼不用處理時間 —— 它比較簡單？ 因為結果取決於 job 什麼時候跑：重播或落後的 partition 都會改變計數。處理時間適合營運訊號 （「最近一分鐘流量裡的錯誤」），不適合業務指標。

Q：如果某個 Kafka partition 沒有收到資料，視窗會怎樣？ 它的 watermark 不會前進，運算子的最小 watermark 就卡住，視窗不再關閉 —— 除非那個來源 被標成閒置。

Q：Exactly-once 是指每個事件只被處理一次嗎？ 不是。故障後事件可能被處理不只一次；exactly-once 指的是狀態與已提交的輸出對每個事件只 反映一次。

Q：要怎麼改變一個正在跑的有狀態 job 的彙總邏輯？ 狀態相容就用 savepoint 再還原。不相容就讓新版本從較早的 offset 寫到新的輸出，驗證後再把 讀取者切過去。

Q：2 秒的 SLA，用 micro-batch 還是 continuous？ 兩者都能達到；短 trigger 的 micro-batch 往往就夠了。SLA 是次秒級、或邏輯需要逐事件的計時器時， 才選逐筆處理。

Q：怎麼測試串流 job？ 用測試 harness 對函式做單元測試（Flink KeyedOneInputStreamOperatorTestHarness、 Kafka Streams TopologyTestDriver），刻意餵入亂序與遲到的事件，再對嵌入式或容器化的 Kafka 跑整合測試。