Message Queues & Streaming

What You Should Understand

• 能解釋 message queue 解決什麼問題：decoupling、buffering、async processing、load leveling
• 能畫出 Kafka 的架構：topics、partitions、offsets、consumer groups、replication
• 知道 Kafka 和 RabbitMQ 的設計哲學差異，以及各自適合的場景
• 能比較 at-most-once、at-least-once、exactly-once，並設計 idempotent consumer
• 理解 stream processing 的 windowing（tumbling/sliding/session）和 watermark 概念
• 能把 streaming 概念接到 ML：real-time features、CDC、event-driven feature store

Why Message Queues

A message queue is an intermediary buffer between producers（寫入事件的服務）and consumers（處理事件的服務）。Without it, services must call each other synchronously — 任何一方變慢或掛掉，整條鏈都跟著出事。

Four Core Benefits

直覺：queue 就像水庫 — 上游暴雨（traffic spike）時先蓄水，下游用固定流量放水。沒有水庫，一場大雨就淹掉下游所有系統。

Queue vs Pub/Sub Models

Kafka 有趣的地方是同時支援兩種模型：同一個 consumer group 內是 queue（partition 分工），不同 consumer group 之間是 pub/sub（各自獨立讀整個 topic）。

Apache Kafka Deep Dive

Kafka is not a traditional queue — it is a distributed, partitioned, replicated commit log。Messages are appended to the end of a log and never modified; consumers read at their own pace by tracking offsets.

Core Concepts

Consumer Groups and Parallelism

一個 topic 有 N 個 partitions，一個 consumer group 內最多 N 個 consumers 能同時工作（多的會 idle）。這是 Kafka 的核心 scaling 機制：

當 consumer 加入或離開時觸發 rebalance — partitions 重新分配。Rebalance 期間該 group 暫停消費，所以頻繁的 consumer crash 會嚴重影響 throughput。

Consumer lag 是最重要的 streaming 監控指標：

Lag 持續上升代表 consumer 跟不上 producer — 需要加 partition/consumer 或優化處理邏輯。

Replication: Leader and ISR

Each partition is replicated across brokers（typical replication factor = 3）:

Producer 的 acks 設定決定 durability：

Why Kafka Is Fast

面試常考「Kafka 為什麼能做到每秒百萬級 messages？」四個關鍵：

Retention and Log Compaction

Kafka 保留 messages 的兩種策略：

Log compaction 讓 Kafka 可以當作「最新狀態的 key-value snapshot」— 新的 consumer 從頭讀 compacted topic 就能重建每個 key 的當前狀態，這正是 streaming feature pipeline 重建 online store 的基礎。

RabbitMQ & Traditional Brokers

RabbitMQ is a smart broker: routing logic lives in the broker, and messages are removed once acknowledged. Producers publish to an exchange, the exchange routes to queues based on bindings, and consumers ack each message.

Exchange Types

Acks, Prefetch, and Redelivery

• Consumer 處理完 message 後送 ack；broker 才刪除該 message
• 處理失敗送 nack（可以 requeue 或丟到 dead letter exchange）
• Prefetch count 限制 unacked messages 數量，避免一個 consumer 抓走太多工作 — 這就是 RabbitMQ 的 backpressure 機制

Smart Broker vs Dumb Broker

Kafka vs RabbitMQ

直覺：Kafka 是「事件的資料庫」，RabbitMQ 是「工作的郵局」。要重播歷史事件、餵多個 downstream（training + monitoring + feature store）→ Kafka。要把任務可靠地派給 worker 處理完就丟 → RabbitMQ。

Cloud Messaging Options

雲端面試（尤其 AWS 生態）常要求你在 managed services 之間選擇：

決策捷徑：

• 單一 consumer 的任務佇列 → SQS
• 一個事件要通知多個系統 → SNS（常見 pattern：SNS fan-out 到多個 SQS）
• 需要根據事件內容做規則路由、串接第三方 SaaS → EventBridge
• 高吞吐 event stream、需要 replay 和多個 analytics consumers → Kinesis（或 self-managed / MSK Kafka）

Delivery Semantics & Reliability

Three Delivery Guarantees

直覺：問題出在「處理」和「記錄進度」是兩個動作，中間隨時可能 crash。先記錄再處理 → 可能漏掉（at-most-once）；先處理再記錄 → 可能重做（at-least-once）。Exactly-once 需要把兩者綁成 atomic 操作。

Exactly-Once in Practice

Kafka 提供 idempotent producer（去除 broker 端的重複寫入）和 transactions（consume-process-produce 綁成一個 atomic unit，Kafka Streams 的 exactly-once 就靠這個）。但一旦 side effect 離開 Kafka（寫外部 DB、呼叫 API），端到端的 exactly-once 就必須靠 idempotent consumer：

Ordering Guarantees

• Kafka：per-partition ordering — 用 entity ID（user_id、account_id）當 key，同一 entity 的事件必然有序
• Retry 會破壞順序：message A 失敗重試時 message B 已經處理完。需要嚴格順序時，per-key sequence number + consumer 端排序，或用 pause-and-retry 而非 skip-and-retry

Dead Letter Queues and Retry Patterns

沒有 DLQ 的下場：一筆壞訊息卡住整個 partition，consumer 無限重試，lag 暴增 — 這叫 poison pill，是 streaming pipeline 最經典的事故。

Batch vs Stream Processing

Comparison

Lambda vs Kappa Architecture

對 ML 的意義：lambda 架構下 batch features 和 streaming features 是兩套 code path — 這是 train-serving skew 的溫床。Kappa（或 feature store 統一 feature 定義）能讓 training 和 serving 讀到一致的 feature 邏輯。

Windowing

Stream 是無限的，聚合（count、sum、mean）必須切出有限的 window：

Feature engineering 對應：詐欺偵測的「過去 5 分鐘刷卡次數」是 sliding window feature；推薦系統的「本次 session 瀏覽類別」是 session window feature。

Event Time, Watermarks, and Late Data

兩種時間概念：

• Event time：事件實際發生的時間（手機 app 產生 click 的時刻）
• Processing time：事件到達 stream processor 的時間

Mobile 斷網、network 延遲會讓事件亂序晚到。Watermark 是系統對「event time 進展到哪了」的估計：

Watermark 越過 window 結束時間 → window 關閉、輸出聚合結果。之後才到的 late events 有三種處理：丟棄、更新已輸出的結果（retraction/upsert）、或送 side output 另外處理。

Tradeoff：allowed lateness 越大 → 結果越完整但延遲越高、state 越大。詐欺偵測寧可快而略不完整；billing 報表寧可慢而精確。

Event-Driven Patterns for Data

Change Data Capture (CDC)

CDC turns database changes into an event stream — ML pipeline 最常見的資料來源之一：

典型鏈路：OLTP DB → Debezium 讀 binlog → Kafka topic（log compacted, key = primary key）→ 同步進 data warehouse（training data）和 online feature store（serving）— 兩邊來自同一個 stream，天然一致。

Event Sourcing (Preview)

傳統 CRUD 只存「當前狀態」；event sourcing 存「所有導致狀態的事件」，狀態是 replay events 的結果。好處：完整 audit trail、可重建任何時間點的狀態（做 point-in-time correct 的 training data 非常有用）、天然適合下游多消費者。代價：查詢當前狀態要靠 materialized views，schema evolution 麻煩。Kafka 的 log + compaction 常被拿來當 event sourcing 的儲存層。

Streaming into Feature Stores

Real-time feature pipeline 的標準形狀：

1. User events 進 Kafka（key = user_id → per-user ordering）
2. Stream processor（Flink / Kafka Streams）做 windowed aggregation：clicks_last_10min、txn_count_last_hour
3. 結果 upsert 進 online store（Redis/DynamoDB, ms-level lookup）同時 append 進 offline store（point-in-time correct training data）
4. Model serving 時從 online store 讀 features

關鍵設計點：同一份 aggregation 邏輯同時餵 online 和 offline → 消除 train-serving skew；upsert 寫法讓 at-least-once 重送不會算錯。

Real-World Use Cases

Case 1: 信用卡詐欺偵測的即時事件流

Scenario：你負責設計詐欺偵測系統的 data pipeline。刷卡交易必須在 100ms 內完成判斷，而最有預測力的 features 是「這張卡過去 5 分鐘 / 1 小時的行為」— 這種 feature 沒辦法在 request 當下從 OLTP DB 現算。

Design：交易事件寫進 Kafka topic（partition key = card_id，保證同卡事件有序）。Flink job 用 sliding window 維護 txn_count_5min、amount_sum_1h、distinct_merchants_1h，持續 upsert 進 Redis online store。Scoring 時 API 從 Redis 拿 pre-computed features（個位數 ms）+ 當筆交易的 real-time features，餵進 model。同一條 stream 也 sink 到 warehouse 做 training data。

• Delivery semantics：feature 計算用 at-least-once + idempotent upsert（重算同一 window 得到相同值）
• Consumer lag 是關鍵 alert — lag 上升代表 features 過期，model 在用舊資料判斷
• Retention 設 7 天以上 → 出事時可以 replay 重建 features

Interview follow-ups：

• 為什麼用 card_id 當 partition key？如果某張卡是超高頻商戶卡造成 hot partition 怎麼辦？
• Flink job 掛掉 10 分鐘後重啟，features 會發生什麼事？怎麼設計 recovery？
• 如果 fraud model 需要「過去 30 天」的 feature，你還會用 streaming 算嗎？（提示：batch + streaming hybrid）

Case 2: A/B Testing 事件收集 Pipeline

Scenario：實驗平台每天收數億筆 exposure 和 conversion events，metric 計算的正確性直接決定實驗結論。Client SDK 在斷網時會 retry，同一個 event 可能送達多次 — 如果不去重，treatment group 的 conversion 會被高估。

Design：SDK 為每個 event 生成 UUID（event_id）＋ client timestamp。Events 進 Kafka（at-least-once ingestion，寧可重複不可漏掉）。下游兩層防線：（1）streaming dedup — consumer 用 Redis 記錄近 24 小時的 event_id，重複直接丟；（2）batch 兜底 — warehouse 每天以 event_id 做 deduplication 重算 metrics。Late events（手機隔天連網才上傳）用 event time 歸屬到正確的實驗日期，並設定 metric 的結算 watermark（例如實驗結束後多等 48 小時才 finalize）。

Interview follow-ups：

• 為什麼 exposure event 掉了比 conversion event 重複更危險？（分母錯 → bias 方向分析）
• Streaming dedup 的 Redis set 只保 24 小時，超過的重複怎麼辦？兩層防線各自的角色？
• Client timestamp 可能被使用者手動調錯，你信 event time 還是 server arrival time？

Case 3: 推薦系統的 User Event Stream 進 Online Feature Store

Scenario：推薦系統的 ranking model 需要 session 級的即時訊號：本次 session 看過哪些類別、最近 20 次點擊的 item embeddings 平均。這些 features 幾秒內就要反映使用者的新行為，否則使用者剛看完的商品類別完全不影響下一頁推薦 — 體感就是「推薦很笨」。

Design：click/view/add-to-cart events 進 Kafka（key = user_id）。Kafka Streams job 維護兩種 state：（1）session window 聚合出 session_category_counts（inactivity gap 30 分鐘）；（2）rolling list of last-20 clicked item IDs → 查 embedding table 算平均向量。兩者 upsert 進 online store。Ranking service 在 request 時讀取。另外同一 topic 有第二個 consumer group 把 raw events 落地 warehouse — pub/sub 模型讓 serving 和 training 各讀各的，互不干擾。

• Compacted changelog topic 備份 streaming state → job 重啟時從 changelog 恢復，不用重算全歷史
• Feature freshness SLA（例如 p99 5 秒內反映）成為 pipeline 的核心監控指標

Interview follow-ups：

• 新用戶（cold start）在 online store 沒有 features，serving 端 fallback 策略是什麼？
• Rebalance 期間 features 更新暫停幾秒，對推薦品質的實際影響？值得上 standby replicas 嗎？
• 為什麼 training 要用 point-in-time 的 feature 值而不是 warehouse 裡的最新值？（feature leakage）

Hands-on: Kafka Patterns in Python

Producer and Consumer with Manual Commit

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer: key-based partitioning keeps per-user ordering
producer = KafkaProducer(
    bootstrap_servers=["localhost:9092"],
    key_serializer=lambda k: k.encode("utf-8"),
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    acks="all",       # wait for all in-sync replicas (durability)
    retries=3,        # transient network errors are retried
)

event = {"event_id": "e-001", "user_id": "u123", "action": "click", "ts": 1720000000}
producer.send("user-events", key=event["user_id"], value=event)
producer.flush()      # block until buffered messages are sent

# Consumer group: partitions are split among consumers with the same group_id
consumer = KafkaConsumer(
    "user-events",
    bootstrap_servers=["localhost:9092"],
    group_id="feature-pipeline",
    enable_auto_commit=False,        # manual commit -> at-least-once
    auto_offset_reset="earliest",    # start from beginning if no committed offset
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
)

for message in consumer:
    handle_event(message.value)      # process FIRST
    consumer.commit()                # commit offset AFTER success
    # crash between the two lines -> event is redelivered (duplicate, not loss)

Idempotent Consumer (Dedup by Event ID)

# At-least-once delivery means duplicates WILL happen.
# Make the side effect idempotent so duplicates are harmless.

processed_ids = set()   # production: Redis SET with TTL, or DB unique constraint

def handle_event(event):
    event_id = event["event_id"]
    if event_id in processed_ids:
        return                        # duplicate from redelivery: skip silently

    # Idempotent write: upsert (SET), not increment (ADD)
    feature_store.upsert(
        key=event["user_id"],
        field="last_action",
        value=event["action"],
    )
    processed_ids.add(event_id)       # record AFTER the side effect succeeds

Tumbling Window Aggregation with a Watermark

from collections import defaultdict

WINDOW_SIZE = 60          # 1-minute tumbling windows (seconds)
ALLOWED_LATENESS = 10     # tolerate events up to 10s late

# window_start -> user_id -> click count
windows = defaultdict(lambda: defaultdict(int))
max_event_time = 0

def assign_window(event_ts):
    # floor the event time to its window start
    return event_ts - (event_ts % WINDOW_SIZE)

def on_event(event):
    global max_event_time
    w = assign_window(event["ts"])
    windows[w][event["user_id"]] += 1
    max_event_time = max(max_event_time, event["ts"])

    # watermark = latest event time seen, minus allowed lateness
    watermark = max_event_time - ALLOWED_LATENESS
    emit_closed_windows(watermark)

def emit_closed_windows(watermark):
    # a window [w, w + WINDOW_SIZE) closes once the watermark passes its end
    for w in sorted(windows):
        if w + WINDOW_SIZE <= watermark:
            counts = windows.pop(w)
            for user_id, n in counts.items():
                # upsert keeps this idempotent under replays
                feature_store.upsert(user_id, f"clicks_1min_{w}", n)
        else:
            break   # windows are sorted; later ones are still open

Interview Signals

What interviewers listen for:

• 你會先問「幾個 consumer？需不需要 replay？」再選 queue vs streaming log，而不是直接喊 Kafka
• 你能解釋 partition 是 ordering 和 parallelism 的單位，並正確選 partition key
• 你主動指出 at-least-once 是常態，並用 idempotency（dedup / upsert）處理 duplicates
• 你會提到 consumer lag、DLQ、poison pill — 代表你維運過真的 streaming pipeline
• 你能把 streaming 接回 ML：real-time features、event time vs processing time、point-in-time correctness

Practice

Flashcards

Quiz