Apache Flink — 資料工程面試深入解析

範圍 — 資料工程面試會探的 Flink 引擎本身：runtime（JobManager、TaskManager、slot、chaining）、部署、DataStream 與 Table / SQL API、程式碼裡的 watermark 與 timer、state backend、checkpoint 與 savepoint 設定、搭配 Kafka 的 exactly-once、Flink SQL 的 join 與 CDC、背壓、記憶體、維運與常見故障 —— 以 Flink 1.20 為準，並標出 Flink 2.x 的差異。與引擎無關的串流理論（為什麼要串流、視窗種類、遲到資料策略、引擎比較）屬於 stream_pipeline.md。 另見：../faq_DE_interview.md — 橫跨所有 DE 主題的總覽與考前檢查清單；./stream_pipeline.md — 把時間語意、視窗、狀態與 exactly-once 當成概念來講；./kafka.md — Flink 通常讀取與寫入的那份 log；./spark.md — 最常拿來和它比較的批次引擎；../flink/faq_flink.md — 較早、附圖的 Flink 筆記。

每一題都先給你會說出口的那句話，再講機制，然後是面試官下一步會追問的取捨。設定名稱以 Flink 1.20 為準；1.20 改過名的 key，舊名稱在 1.x 仍然有效，並會註明。

1) Flink 是什麼、用在哪裡 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

1-1) 一段話的答案

Flink 是分散式串流處理器：它執行一個由 operator 組成的資料流，一次處理一個事件，把每個 key 的狀態放在 operator 旁邊，並用週期性的分散式快照（稱為 checkpoint）讓狀態具備容錯能力。

• 串流優先 —— 批次 job 就是一個有界的串流；同一個 runtime 兩者都跑。
• 有狀態 —— 狀態是 job 中一等公民、分區、會做 checkpoint 的一部分，在 RocksDB backend 上可以長到 TB 級。
• 事件時間 —— watermark、事件時間 timer 與遲到資料處理都是內建的，所以結果取決於事件 何時發生，而不是何時抵達。
• 狀態 exactly-once —— 故障之後，每個 operator 的狀態都還原到同一個一致的 checkpoint， 來源則倒回對應的 offset。

1-2) Flink vs Spark

Flink 逐筆處理，延遲是毫秒級，並管理大量狀態；Spark Structured Streaming 每次 trigger 跑一個 micro-batch，當團隊本來就用 Spark 做批次時比較合適。

面試時說的那句話：「延遲、事件時間或狀態大小是難點時用 Flink；難點是批次量、串流只是旁邊 幾個聚合時用 Spark。」 四個引擎的比較表（含 Kafka Streams 與 Beam）在 stream_pipeline.md §9-1。

1-3) 什麼時候不要用 Flink

• 分鐘級的新鮮度就夠了 —— 排程批次 job 或 micro-batch 執行起來、除錯起來都更便宜。
• Kafka 進、Kafka 出，狀態很小 —— Kafka Streams 跑在服務裡面，沒有叢集要維運。
• 沒有人能維運它 —— Flink job 是長時間運行的分散式系統：checkpoint、savepoint、升級與 on-call。改用託管服務（Amazon Managed Service for Apache Flink、Confluent Cloud for Apache Flink、Ververica）或更簡單的工具。

2) 架構與 Runtime ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

2-1) JobManager、TaskManager 與 slot

JobManager 規劃並協調 job，TaskManager 執行它，每個 TaskManager 提供固定數量的 task slot —— slot 是 job 的一個平行切片被排進去的單位。

                ┌─────────────── JobManager ───────────────┐
                │ Dispatcher · ResourceManager · JobMaster │
                │ checkpoint coordinator · scheduler       │
                └───────┬──────────────────────┬───────────┘
       deploy subtasks  │                      │  trigger checkpoints
            ┌───────────▼──────────┐   ┌───────▼──────────────┐
            │ TaskManager 1        │   │ TaskManager 2        │
            │ ┌──────┐ ┌──────┐    │   │ ┌──────┐ ┌──────┐    │
            │ │slot 1│ │slot 2│    │   │ │slot 1│ │slot 2│    │
            │ └──────┘ └──────┘    │   │ └──────┘ └──────┘    │
            │ state · net buffers  │◄─►│ state · net buffers  │
            └──────────────────────┘   └──────────────────────┘

• taskmanager.numberOfTaskSlots（預設 1）設定每個 TaskManager 的 slot 數；常見做法是每個 CPU 核心一個 slot。
• JobManager 是唯一的協調者；正式環境會開 high availability（ZooKeeper 或 Kubernetes HA 服務），讓新的 JobManager 接手，並從最近一次完成的 checkpoint 還原 job。

2-2) 從程式到執行中的 task

你的程式碼先變成 StreamGraph，chain operator 之後最佳化成 JobGraph，JobManager 再把它 展開成 ExecutionGraph，每個平行 subtask 一個頂點。

StreamGraph      every operator in your program
    │ chain operators that can run in one thread
JobGraph         what is submitted: chained vertices + data exchanges
    │ expand each vertex by its parallelism
ExecutionGraph   one execution vertex per subtask, kept by the JobMaster
    │ deploy
physical tasks   subtasks running in TaskManager slots

Operator chaining 會在連線是一對一（forward）且平行度相同時，把相鄰的 operator 合併成 一個 task —— source → map → filter 變成一個執行緒，中間沒有序列化也沒有網路跳轉。keyBy、 rebalance 或平行度改變會打斷 chain。disableChaining() / startNewChain() 用來把一個很重的 operator 拆出去，讓 Web UI 看得出是哪一個慢。

2-3) 平行度、slot 與 slot sharing

預設情況下，一個 slot 可以放 job 中每一個 operator 各一個 subtask，所以 job 需要的 slot 數等於它最高的 operator 平行度 —— 不是總和。

job: source(p=4) → keyBy → window(p=4) → sink(p=2)

slot sharing on (default):          4 slots
  slot 1: source-1, window-1, sink-1
  slot 2: source-2, window-2, sink-2
  slot 3: source-3, window-3
  slot 4: source-4, window-4

• 平行度可以每個 operator 設（.setParallelism(n)）、每個 job 設（-p、parallelism.default）， 或完全不設（預設 1）。每個 operator 的設定優先。
• 讓來源對齊 Kafka 分區數：來源平行度高於分區數時，會有 subtask 閒置 —— 而閒置的 subtask 會拖住 watermark（§4-2）。
• Max parallelism（pipeline.max-parallelism）是 key group 的數量 —— 也就是你將來能擴到 的上限。在第一次正式部署前就設定好（§5-4）。

2-4) 部署模式

正式環境的 job 跑在 application mode —— 每個 job 一個叢集、main() 在 JobManager 上執行 —— 通常透過 Flink Kubernetes Operator 跑在 Kubernetes 上。

資源提供者：Kubernetes（原生或 Operator）、YARN 或 standalone。用 Operator 時，一個 job 就是一個 Kubernetes 資源：

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: order-agg
spec:
  image: registry.example.com/order-agg:1.4.0
  flinkVersion: v1_20
  serviceAccount: flink
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    state.backend.type: rocksdb
    execution.checkpointing.interval: "60s"
    execution.checkpointing.incremental: "true"
    execution.checkpointing.dir: s3://flink/checkpoints/order-agg
  jobManager:
    resource: { memory: "2048m", cpu: 1 }
  taskManager:
    resource: { memory: "8192m", cpu: 4 }
  job:
    jarURI: local:///opt/flink/usrlib/order-agg.jar
    parallelism: 16
    upgradeMode: savepoint     # on a spec change: stop with a savepoint, redeploy, restore

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

3-1) 比較

預設用 SQL / Table API；需要自訂狀態、timer 或 planner 表達不了的控制時，才降到 DataStream API。

• PyFlink 包裝了 Table API 與 DataStream API；Python UDF 在另一個 Python 行程裡執行， 所以熱路徑上優先用內建函式，和 PySpark 一樣。
• Scala：Scala API 在 2.0 移除；Scala 使用者改呼叫 Java API。
• API 可以混用：Table 能轉成 DataStream 再轉回來（toChangelogStream、 fromChangelogStream）。

3-2) 一個最小的 DataStream job

一個 job 就是：environment → 帶 watermark 策略的來源 → 轉換 → sink → execute()。每個 有狀態的 operator 都要給 uid。

// java
// Per-customer order totals over 1-minute event-time windows, Kafka → Kafka.
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60_000);   // every 60 s; exactly-once is the default mode

KafkaSource<Order> source = KafkaSource.<Order>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("orders")
    .setGroupId("order-agg")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    .setValueOnlyDeserializer(new OrderDeserializer())
    .build();

WatermarkStrategy<Order> watermarks = WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(10))   // events up to 10 s late are on time
    .withTimestampAssigner((order, kafkaTs) -> order.eventTimeMillis)
    .withIdleness(Duration.ofMinutes(1));                       // an empty partition stops holding the watermark

env.fromSource(source, watermarks, "orders").uid("orders-source")
    .keyBy(order -> order.customerId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .aggregate(new SumAmount())          // incremental: one accumulator per key and window
    .uid("customer-1m-total")
    .sinkTo(totalsSink).uid("totals-sink");

env.execute("order-agg");

• aggregate / reduce 每個視窗只保留一個 accumulator；視窗上的 process 會把每個元素都 暫存到視窗觸發為止 —— 只有邏輯真的需要全部元素時才用。
• 以 Duration 為參數的視窗與 lateness 方法是 1.19+；更早的版本用 Time.minutes(1)。

3-3) KeyedProcessFunction 與 timer

KeyedProcessFunction 會帶著某個 key 的狀態看到該 key 的每個事件，並能設定 timer，在某個 事件時間或處理時間回呼它 —— 「如果 N 分鐘內 X 沒有發生」就是這樣表達的。

// java
// Alert when an order is not paid within 15 minutes of being created (event time).
public class UnpaidOrderAlert extends KeyedProcessFunction<String, OrderEvent, String> {
    private transient ValueState<Long> deadline;   // when the pending timer fires; null = none

    @Override
    public void open(OpenContext ctx) {            // 1.19+; open(Configuration) before, removed in 2.0
        deadline = getRuntimeContext().getState(
            new ValueStateDescriptor<>("deadline", Long.class));
    }

    @Override
    public void processElement(OrderEvent e, Context ctx, Collector<String> out) throws Exception {
        if (e.type == EventType.CREATED) {
            long fireAt = e.eventTimeMillis + Duration.ofMinutes(15).toMillis();
            deadline.update(fireAt);
            ctx.timerService().registerEventTimeTimer(fireAt);
        } else if (e.type == EventType.PAID && deadline.value() != null) {
            ctx.timerService().deleteEventTimeTimer(deadline.value());
            deadline.clear();                      // paid in time: nothing to alert
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        out.collect("order " + ctx.getCurrentKey() + " unpaid after 15 minutes");
        deadline.clear();
    }
}

• 事件時間 timer 在 watermark 越過它時觸發，所以從 checkpoint 或回補重播時也正確； 處理時間 timer 依牆上時鐘觸發。
• Timer 以 key 區分、同一個 key 與時間戳只會有一個，並存放在 checkpoint 裡。
• 陷阱：PAID 比 CREATED 先到（亂序）會留下一個本來不該觸發的警示 —— 提早到的 PAID 也要存進狀態；而一個永遠等不到結束事件的 key，若沒有 timer 或 TTL 清掉，狀態會永遠留著。

3-4) Async I/O 與 side output

向外部服務補資料要用 async I/O，絕不要在 map 裡做阻塞呼叫；不屬於主輸出的資料送到 side output，而不是丟掉。

// java
// Up to 100 lookups in flight per subtask; a lookup slower than 500 ms times out.
DataStream<EnrichedOrder> enriched = AsyncDataStream.unorderedWait(
    orders, new CustomerLookup(), 500, TimeUnit.MILLISECONDS, 100);

// Late events go to a side output instead of being silently dropped.
OutputTag<Order> lateTag = new OutputTag<Order>("late-orders") {};
SingleOutputStreamOperator<Total> totals = orders
    .keyBy(order -> order.customerId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .allowedLateness(Duration.ofMinutes(5))
    .sideOutputLateData(lateTag)
    .aggregate(new SumAmount());
DataStream<Order> lateOrders = totals.getSideOutput(lateTag);

• 在 map 裡做阻塞呼叫，整個 subtask 都得等網路 —— 吞吐量崩掉，背壓往上游擴散。
• unorderedWait 依完成順序輸出結果（只在 watermark 之間保持順序）；orderedWait 保持輸入 順序，代價是延遲。
• Async I/O 的狀態會進 checkpoint：還原後，進行中的請求會重送，所以查詢必須可以安全重複。

4) Flink 裡的 Watermark 與遲到資料 ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

概念 —— 事件時間 vs 處理時間、watermark 承諾了什麼、遲到資料的策略 —— 在 stream_pipeline.md §3 與 §4-3。這一節講 Flink 怎麼實作它們。

4-1) Watermark 從哪裡來

Watermark 在來源產生，每個來源 split（每個 Kafka 分區）各自一個，而 operator 的 watermark 是它所有輸入 watermark 的最小值。

• forBoundedOutOfOrderness(d) 送出 看過的最大事件時間 − d − 1 ms； forMonotonousTimestamps() 是 d = 0 的特例。
• 它們是週期性送出的，每 pipeline.auto-watermark-interval（預設 200 ms）一次，不是每筆都送。
• 在來源裡產生 watermark（fromSource(..., strategy, ...)），不要在來源之後用 assignTimestampsAndWatermarks：只有在來源內部才可能做到每個分區各自的 watermark，後面的 assigner 看到的分區已經交錯在一起了。
• 在 SQL 裡，同一件事是一個欄位宣告： WATERMARK FOR order_time AS order_time - INTERVAL '10' SECOND。

4-2) 閒置分區與 watermark 對齊

一個沒有資料的分區會拖住所有下游的 watermark，一個遠遠跑在前面的分區會讓狀態變大 —— withIdleness 解決前者，watermark 對齊解決後者。

Idleness 有代價：閒置分區醒來時，它最早的那些事件可能已經算遲到了。

4-3) DataStream API 裡的遲到事件

視窗在 watermark 越過它的結束時間時觸發；allowedLateness 讓它繼續開著接受更新，再更晚的 仍然送到 side output。

allowedLateness 會讓每個視窗的狀態多留那麼久，所以記憶體成本和 key 的數量成正比。

5) 狀態與 State Backend ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

5-1) 狀態的種類

Keyed state 在 keyBy 之後以目前的 key 為範圍，幾乎所有 job 都用它；operator state 屬於 一個平行 subtask，broadcast state 則是每個 subtask 上一模一樣的一份。

Keyed vs operator state 與 broadcast 模式的深入內容在 stream_pipeline.md §5-1 與 §7-4。

5-2) HashMap vs RocksDB

HashMap 把狀態當成 Java 物件放在 heap 上 —— 最快，但受記憶體限制；RocksDB 把它序列化放在 本機磁碟 —— 每次存取比較慢，但大小由磁碟決定，也是唯一能做 incremental checkpoint 的選擇。

• 用 state.backend.type: hashmap | rocksdb 設定。Checkpoint 存到哪裡是另一件事： execution.checkpointing.dir（1.20 之前是 state.checkpoints.dir），正式環境一定是耐久儲存 （S3、GCS、HDFS）。
• 在 RocksDB 上，狀態的值要小，並用 MapState 取代一個很大的 ValueState<Map> —— 每次 存取都會反序列化整個值。
• Flink 2.0 新增 ForSt，一個實驗性的分離式 backend，把狀態放在遠端儲存，讓 TaskManager 的 本機磁碟不再是上限。

5-3) 讓狀態有界：TTL

每一份 keyed state 都需要一條死路 —— 一個會清掉它的 timer，或一個 TTL —— 否則 job 的狀態 會隨著它看過的每一個 key 一直長。

// java
StateTtlConfig ttl = StateTtlConfig
    .newBuilder(Duration.ofHours(24))                               // 1.x before 1.19: Time.hours(24)
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)      // reading does not extend it
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();

ValueStateDescriptor<Long> lastSeen = new ValueStateDescriptor<>("last-seen", Long.class);
lastSeen.enableTimeToLive(ttl);

• TTL 只用處理時間，所以回補時狀態不會像即時流量那樣過期。
• 過期的 entry 是延遲清除的：在存取時、在完整快照時，或由 RocksDB compaction 清掉 —— 記憶體 不會在 TTL 一到就降下來。
• 在 SQL 裡，table.exec.state.ttl（或 1.18+ 每個 operator 的 STATE_TTL hint）限制 join 與 聚合的狀態；沒有它，regular join 會永遠保留兩邊（§8-3）。

5-4) Key group 與重新擴縮

Keyed state 被切成固定數量的 key group —— 也就是 max parallelism —— 重新擴縮時是把整個 key group 在 subtask 之間搬動；這就是為什麼 max parallelism 一改就會失去狀態。

max parallelism 128  →  128 key groups;  key → hash(key) % 128

parallelism 4:   subtask 0: groups 0-31   subtask 1: 32-63   subtask 2: 64-95   subtask 3: 96-127
rescale to 8:    each subtask takes 16 groups — state moves by whole group, no key is rehashed

• 預設的 max parallelism 由第一次部署的平行度推出來（至少 128、最多 32768），所以用不同平行度 第一次部署的 job 可能得到不同的值。明確設定 pipeline.max-parallelism。
• 平行度不可能超過 max parallelism；max parallelism 遠大於平行度只會讓每個 key group 多一點 額外成本。常見做法是取一個 2 的冪次，至少是預期尖峰的好幾倍。

6) Checkpoint、Savepoint 與復原 ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

6-1) Checkpoint 是什麼

Checkpoint 是每個 operator 的狀態加上來源 offset 的一致快照，而且不需要停下 job： JobManager 在來源注入 barrier，每個 operator 在 barrier 通過時對自己的狀態做快照。

故障時，Flink 重啟受影響的 task，從最近一次完成的 checkpoint 還原狀態，並把來源倒回它的 offset —— 所以狀態對每一筆資料恰好反映一次，即使有些資料被處理了兩次。Barrier 機制、 aligned vs unaligned checkpoint，以及 checkpoint vs savepoint 在 stream_pipeline.md §6-1 到 §6-3。

對 Kafka 來源而言，重要的 offset 是 checkpoint 裡的那些。Flink 只在 checkpoint 完成時把 offset commit 回 Kafka，而且只是為了讓 lag 儀表板能運作；復原時從來不讀它們。

6-2) 值得記住名字的 checkpoint 設定

6-3) Savepoint 與升級 job

Savepoint 是你觸發、由你擁有的 checkpoint：用它停掉 job、部署新的 jar、從 savepoint 啟動 —— 只要每個有狀態的 operator 都保持同一個 uid，狀態就會帶過去。

# Stop with a savepoint (drains in-flight data cleanly), then resume the new version from it.
./bin/flink stop --savepointPath s3://flink/savepoints/order-agg <jobId>
./bin/flink run -d -s s3://flink/savepoints/order-agg/savepoint-3f2a1b-6c8d1e0f9a7b target/order-agg-1.5.jar

要說出口的規則：「每個有狀態的 operator 都要有 uid()，從第一次部署開始。」 Savepoint 可以在不同 backend 之間移植（canonical 格式）；checkpoint 不保證可以。

6-4) 重啟、failover region 與 HA

Task 失敗時，Flink 只重啟透過管線化資料和它相連的那些 task —— 也就是 failover region —— 並從最近的 checkpoint 還原它們，依 restart strategy 重試。

重啟迴圈 —— job 還原、在同一筆資料上失敗、再還原 —— 會表現為一直上升的 numRestarts 指標。 這需要修程式（或加一個 dead-letter 輸出），而不是更多重試。

7) 搭配 Kafka 的端到端 Exactly-Once ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

7-1) Flink 保證什麼、sink 補上什麼

Flink 的 checkpoint 讓狀態 exactly-once；輸出要 exactly-once，sink 必須是冪等的，或在 checkpoint 時以交易方式 commit。

兩階段提交如何和 checkpoint 對齊在 stream_pipeline.md §6-5。

7-2) 設定成 exactly-once 的 Kafka sink

// java
KafkaSink<String> totalsSink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("order-totals")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("order-agg")          // unique per job on the cluster
    .setProperty("transaction.timeout.ms", "900000") // 15 min: must not exceed the broker's max
    .build();

以下每個陷阱面試官都會問：

• 交易逾時。 Sink 預設的 transaction.timeout.ms 是 1 小時；broker 的 transaction.max.timeout.ms 預設是 15 分鐘，所以在改掉其中一個之前，job 一啟動就失敗。這個 逾時也必須大於預期最長的中斷時間加上 checkpoint 間隔，否則 broker 會中止一個 Flink 還打算 commit 的交易 —— 資料遺失。
• 讀取端必須用 isolation.level=read_committed。 預設（read_uncommitted）的 consumer 會看到已中止交易的資料 —— 正是你花了成本想避免的重複。
• 延遲等於 checkpoint 間隔。 下游讀取端只在交易 commit 時才看到它的資料，也就是在一次 checkpoint 結束時；間隔 1 分鐘，輸出就以 1 分鐘為一步抵達。
• 每個 job 的 transactionalIdPrefix 要唯一。 兩個 job 共用同一個 prefix，會互相 fence 掉對方的 producer。

7-3) 其他 sink

• File sink（FileSink） —— 寫入 in-progress 的 part 檔，在 checkpoint 完成時把它們轉成 finished；搭配 bulk 格式（Parquet）時每次 checkpoint 都會滾一個新檔，所以間隔短就會產生很多 小檔。
• Lakehouse 資料表 —— Iceberg、Paimon、Delta 與 Hudi 的 Flink sink 每次 checkpoint commit 一個資料表 snapshot；新鮮度和檔案數之間是同樣的取捨。
• JDBC —— 用以主鍵為 key 的 upsert 語句，讓重播是覆寫而不是重複；JDBC sink 也有 XA exactly-once 模式，但很少值得它的維運成本。

8) Flink SQL 與 Table API ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

8-1) 一段話講 dynamic table

對串流下的 Flink SQL 查詢，是對一張一直在變的表做的持續查詢；它的結果是另一張一直在變的 表，以 changelog 的形式送出。

只有附加的結果（過濾、視窗聚合）可以送到任何 sink；會更新的結果（沒有視窗的 GROUP BY、 regular join）會送出更新與撤回，需要一個有主鍵的 sink。Changelog 的列種類（+I、-U、+U、 -D）在 stream_pipeline.md §9-2， SQL 的視窗聚合在 §10-1。

8-2) 累計總額，從 Kafka 到 upsert-Kafka

CREATE TABLE orders (
  order_id     STRING,
  customer_id  STRING,
  amount       DECIMAL(10, 2),
  order_time   TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time - INTERVAL '10' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'customer-totals',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest',
  'format' = 'json'
);

-- An updating result needs a sink with a primary key.
CREATE TABLE customer_totals (
  customer_id  STRING,
  order_count  BIGINT,
  total_amount DECIMAL(18, 2),
  PRIMARY KEY (customer_id) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'customer-totals',
  'properties.bootstrap.servers' = 'kafka:9092',
  'key.format' = 'json',
  'value.format' = 'json'
);

INSERT INTO customer_totals
SELECT customer_id, COUNT(*), SUM(amount)
FROM orders
GROUP BY customer_id;

• 這個聚合每個顧客保留一列狀態，永遠保留，除非設定 table.exec.state.ttl —— 而有了 TTL， 閒置超過它的顧客會從零重新開始。
• upsert-kafka 把 key 寫成 Kafka 訊息的 key，刪除寫成 tombstone，所以 compacted topic 會 保留每個顧客最新的總額。
• Planner 會在 EXPLAIN CHANGELOG_MODE SELECT … 裡顯示每個 operator 的 changelog 模式。

8-3) 四種 join

依每一邊要保留多久來選 join：regular join 永遠保留兩邊；其他三種都限制了狀態。

串流 join 的概念在 stream_pipeline.md §7。

8-4) Flink CDC

Flink CDC 直接把資料庫的變更 log（MySQL binlog、PostgreSQL logical replication）當成 changelog 串流讀進來 —— 先 snapshot、再讀 log —— 中間不需要 Debezium，也不需要 Kafka。

• 來源以平行的 chunk 做一致的初始 snapshot，不鎖資料表，然後在 snapshot 結束的那個確切位置 切換到 log。
• CDC 表是會更新的表，所以它能像其他表一樣 join 與聚合，並寫進 upsert sink —— 常見模式是 MySQL → Flink CDC → Paimon / Iceberg / Hudi。
• Flink CDC 3.x 新增 YAML pipeline，可以同步整個資料庫，包含 schema 變更。
• 如果中間改放 Kafka（Debezium → Kafka → Flink），用 debezium-json 格式讀 topic —— kafka.md 講的是那一側。

8-5) 調校 SQL 聚合

9) 背壓與效能 ★★★★☆優先度 4／5 — 高價值 — 這裡有缺口就會掉關

9-1) 資料怎麼流動，以及為什麼背壓是安全的

Subtask 之間用網路 buffer 交換資料，採用 credit-based 流量控制：傳送端只送接收端宣告還有 空間的那麼多 buffer，所以慢的 operator 會讓它的上游變慢，而不是把記憶體用光。

• Flink 的背壓是症狀，不是故障：最慢的 operator 決定整條管線的吞吐量，Kafka lag 隨之增加。
• 它會傷害 checkpoint：barrier 排在暫存資料後面，所以 aligned checkpoint 會變慢或逾時。 Unaligned checkpoint（barrier 超車越過 buffer）與 buffer debloating （taskmanager.network.memory.buffer-debloat.enabled，把傳輸中的資料縮到大約一秒的量）是 兩個解法。

9-2) 找出瓶頸

在 Web UI 裡，找出第一個 busy、而它上游全都 backpressured 的 operator —— 那個 operator 就是瓶頸。

接著問它為什麼 busy：只有一個 subtask busy、其他閒置，是傾斜（熱 key —— stream_pipeline.md §8-3）；所有 subtask 都 busy，是平行度 太低或程式碼太貴 —— Web UI 的 flame graph（rest.flamegraph.enabled）會顯示是哪個方法； sink busy 則是外部系統的極限。

9-3) 調校，依你會檢查的順序

1. 是傾斜嗎？ 比較每個 subtask 的資料筆數。用預聚合（two-phase、mini-batch）或加鹽的 key 再做第二次聚合來修。
2. 是阻塞呼叫嗎？ 把同步查詢換成 async I/O（§3-4）或有快取的 lookup join。
3. 是狀態存取嗎？ 在 RocksDB 上，SQL 打開 mini-batch、用 MapState、值保持小、給 RocksDB 更多 managed memory。
4. 是序列化嗎？ 泛型型別退回 Kryo 很慢 —— 改用 POJO、Avro 或註冊過的型別； pipeline.generic-types: false 會讓退回 Kryo 的情況在啟動時就失敗。
5. 是平行度嗎？ 調高 —— 來源最多到 Kafka 分區數，keyed operator 最多到 max parallelism。
6. 是 checkpoint 嗎？ Checkpoint 很久或一直失敗：RocksDB incremental、unaligned checkpoint、拉長間隔、min-pause。

10) 記憶體模型與容量規劃 ★★★☆☆優先度 3／5 — 值得會 — 多半是必備模式的變形

10-1) TaskManager 記憶體

設定總量（taskmanager.memory.process.size），讓 Flink 自己分配；值得知道的是 heap、 managed memory（RocksDB 的快取）與網路 buffer。

• 用 hashmap 狀態時，managed memory 大多用不到 —— 調低比例，把空間讓給 heap。
• 在 Kubernetes 上，container 被殺掉卻沒有 Java OutOfMemoryError，通常代表 RocksDB 或其他 native 配置超過了上限：調高 JVM overhead，或確認 state.backend.rocksdb.memory.managed （預設 true）仍然開著。

10-2) 從工作負載推算規模

Peak input       50,000 events/s, 1 KB each
One subtask      ~5,000 events/s through the heaviest operator (measure it; this is an assumption)
Parallelism      50,000 / 5,000 = 10, plus headroom for catch-up after an outage → 16
Kafka partitions ≥ 16, so every source subtask has a partition
Slots            16 → 4 TaskManagers × 4 slots
Max parallelism  128 — room to grow 8× without losing state
State            keys × bytes per key; above a few GB per TaskManager → RocksDB + incremental checkpoints

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

11-1) 要監控什麼

11-2) 擴縮執行中的 job

• 手動 —— 用 savepoint 停掉，再以新的平行度（≤ max parallelism）從它啟動。
• Adaptive scheduler（jobmanager.scheduler: adaptive）—— job 使用當下可用的 slot；在 standalone application 叢集搭配 reactive mode 時，加 TaskManager 就會讓 job 擴大。
• Kubernetes Operator 的 Autoscaler（job.autoscaler.enabled: "true"）—— 量測每個 operator 的 busy 時間與目標速率，再逐個 operator 重新擴縮；Flink 1.18+ 可以就地套用新的平行度，不必 完整重新部署。
• 來源的擴縮無法超過 Kafka 分區數。

11-3) 重新處理

修完 bug 要重播歷史資料時：啟動一個新的 job，從最早的 offset（或某個時間戳）開始，寫到 新的輸出，讓它追上，把讀取端切過去，再停掉舊的。用修好的程式碼還原舊的 savepoint，只會修到 它之後的事件。這個模式在 stream_pipeline.md §12-3。

12) 常見故障與解法 ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

13) 常見問題 ★★★★★優先度 5／5 — 必備 — 幾乎每一輪面試都會出現

13-1) 快問快答

13-2) 設計題：「即時偵測信用卡詐欺」

題目 —— 刷卡交易以每秒 20,000 筆的速度進入 Kafka；若一張卡在 1 分鐘內刷超過 5 次，或在 10 分鐘內有兩筆交易相距超過 500 公里，就標記它；警示必須在幾秒內送到下游服務。

Kafka: transactions (64 partitions, key = card_id)
   │
Flink job (application mode, Kubernetes Operator, parallelism 32, max parallelism 256)
   ├─ KafkaSource, watermark = event time − 5 s, idleness 1 min
   ├─ keyBy(card_id)
   ├─ KeyedProcessFunction
   │     state: recent transactions (ListState, pruned by event-time timer after 10 min)
   │     rule 1: count in the last 1 min > 5
   │     rule 2: distance / time to the previous transaction too large
   ├─ broadcast state: rule thresholds, updated from a config topic without a redeploy
   └─ sinks: alerts → Kafka (EXACTLY_ONCE, checkpoint every 10 s)
             late events → side output → lake table for audit
   │
Fraud service consumes alerts with read_committed

要講的重點，大致依面試官追問的順序：

• 為什麼用 KeyedProcessFunction 而不是視窗 —— 規則 2 要拿每筆交易和前一筆比，固定視窗 表達不了；10 分鐘的 timer 讓狀態保持有界。
• 延遲 vs exactly-once —— 警示 topic 只看得到已 commit 的資料，所以 checkpoint 間隔（10 秒） 就是延遲的下限。如果幾秒太慢，改用 AT_LEAST_ONCE，在下游依交易 id 去重。
• 狀態大小 —— 活躍卡片數 × 幾筆最近的交易：每個 key 很小，但 key 很多 → RocksDB 加上 incremental checkpoint。
• 熱 key —— 卡號分布很平均；以商家為單位的規則就不是，需要預聚合。
• 改規則 —— 用 broadcast state，讓門檻不需要 savepoint 與重新部署就能改；改邏輯則是一次 savepoint 升級（§6-3）。
• 故障 —— TaskManager 掛掉時從最近的 checkpoint 還原，最多重播 10 秒的輸入；冪等的警示 id 讓任何重複都無害。