Lab — thiết kế một stream pipeline thật

Team bảo mật của VietBank phát hiện mỗi đêm có hàng trăm cảnh báo gian lận bị bỏ sót vì pipeline cũ dùng Spark batch chạy theo giờ — kẻ gian đã rút sạch tài khoản trước khi batch kịp phát hiện. Hệ thống mới cần phát cảnh báo trong vòng 30 giây kể từ khi cửa sổ phát hiện vi phạm: tổng giao dịch của một tài khoản vượt ngưỡng riêng trong bất kỳ cửa sổ 5 phút nào.

Đây là bài tổng hợp toàn module: bạn áp dụng đồng thời tư duy stream vs batch (bài 01), log-based broker và partition key của Kafka (bài 02), windowing và watermark cho event trễ (bài 03), stream-table join và đảm bảo exactly-once (bài 04). fadeLevel: independent có nghĩa bạn phải tự thiết kế trước khi đọc gợi ý.

1. 🎯 Đề bài

Bạn nhận đặc tả hệ thống như sau:

Nguồn dữ liệu:

Đặc tính trễ của từng channel: ATM/POS kết nối ổn định, trễ dưới 5 giây; mobile app buffer khi mất mạng, trễ 2–3 phút.

KTable tham chiếu (từ topic accounts-changelog): account_id, account_name, alert_threshold — ngưỡng cảnh báo riêng từng tài khoản (VIP có ngưỡng cao hơn, tài khoản mới có ngưỡng thấp hơn).

Yêu cầu output: ghi record vào Kafka topic fraud-alerts với schema (account_id, account_name, window_start, window_end, total_amount_vnd, tx_count) khi tổng amount_vnd trong cửa sổ vượt alert_threshold. SLA: ghi vào sink trong vòng 30 giây kể từ khi cửa sổ phát hiện vi phạm.

Bốn việc bạn cần tự làm:

(a) Vẽ pipeline — mỗi operator ghi rõ: đọc input gì, làm một thao tác (filter / project / window / join / sink), output ra đâu. Vẽ cả nhánh xử lý late event.

(b) Chọn window type + watermark delay và lập luận theo SLA cùng đặc tính trễ từng channel: vì sao dùng event-time thay processing-time? Tumbling hay sliding, với duration bao nhiêu? Watermark delay bao nhiêu và trade-off của nó?

(c) Chọn delivery guarantee + cơ chế exactly-once cho fraud-alerts: kịch bản nào gây cảnh báo trùng lặp nếu dùng at-least-once? exactly-once cần hai thành phần nào kết hợp?

(d) Chọn partition key cho topic bank-transactions và lập luận: sai key gây hậu quả gì cho aggregate của window operator?

2. 🔍 I-P-O

3. 📦 Concept mapping

Bài lab này là bằng chứng trực tiếp cho hai learning outcome cuối của module:

• LO3 — Design windowing và lập luận watermark (bài 03): yêu cầu (a) và (b) buộc bạn vẽ pipeline với window operator thật, chọn event-time vs processing-time, và lập luận watermark delay theo SLA + đặc tính từng channel — không chỉ nhớ tên window type.
• LO4 — Exactly-once và fault tolerance (bài 04): yêu cầu (c) buộc bạn xác định kịch bản duplicate và chọn + biện minh cơ chế đảm bảo exactly-once — không chỉ bật một flag.

Hai outcome được gọi lại để tổng hợp:

• LO1 (bài 01 — tư duy stream): batch không dùng được cho bài toán này vì input unbounded và SLA 30 giây; batch chạy theo giờ không thể đáp ứng.
• LO2 (bài 02 — Kafka log/offset): partition key quyết định partition → ảnh hưởng thứ tự và tính đúng đắn của aggregate; Kafka giữ log cho phép consumer replay.

4. ▶️ Starter — tự thiết kế trước

Thay mọi [TODO] bằng câu trả lời của bạn. Comment sau -- là nơi ghi lý luận.

PIPELINE fraud_detection():

  -- ========================================================
  -- SOURCE: Kafka topic bank-transactions
  -- ========================================================
  tx_stream <- kafka_source("bank-transactions")
                .assign_event_time(t -> [TODO: dung truong nao lam event-time?])
                .assign_watermark(max_event_time - [TODO: delay bao nhieu? ly do?])
  -- Tai sao dung event-time thay vi processing-time?
  -- [TODO: viet ly do cu the]

  -- ========================================================
  -- STEP 1: Filter + Project (stateless)
  -- ========================================================
  clean <- tx_stream
             .filter(t -> [TODO: dieu kien hop le?])
             .project(t -> [TODO: chi giu field nao cho buoc aggregate?])
  -- Chieu som de: [TODO: giai thich loi ich]

  -- ========================================================
  -- STEP 2: Windowed Aggregate (stateful)
  -- ========================================================
  windowed <- clean
                .key_by(t -> [TODO: partition key la gi? tai sao?])
                .window([TODO: TumblingWindow(?) / HoppingWindow(?,?) / SlidingWindow(?) / SessionWindow(?)])
                -- Chon window type nay vi: [TODO]
                .aggregate(
                    [TODO: ham aggregate + group key]
                )
  -- Output fields: [TODO: liet ke cac field output]

  -- Late event (delay > watermark_delay):
  -- Xu ly bang: [TODO: allowed_lateness=? / side_output / bo qua]
  -- Ly do: [TODO]

  -- ========================================================
  -- STEP 3: Non-windowed Stream-Table Join
  -- ========================================================
  accounts_tbl <- kafka_source("[TODO: topic nao?]")
                    .as_table(key = "account_id")
  enriched <- windowed.join_table(accounts_tbl, key = "account_id")
  -- Dung non-windowed join (khong phai stream-stream join) vi:
  -- [TODO: giai thich tai sao KTable khong can window]

  -- ========================================================
  -- STEP 4: Filter nguong + Sink
  -- ========================================================
  alerts <- enriched
              .filter(r -> [TODO: dieu kien alert?])

  alerts.sink_to_kafka("fraud-alerts",
                        key     = [TODO: sink key la gi? tai sao?],
                        delivery = [TODO: at_most_once / at_least_once / exactly_once],
                        mechanism = [TODO: tai sao can co ca 2 thanh phan?])
  -- at-least-once gay ra van de gi neu consumer crash roi restart?
  -- [TODO: mo ta kich ban duplicate]

  -- ========================================================
  -- Uoc tinh state cua window operator
  -- ========================================================
  -- State per (account, window) = SUM 8 + COUNT 4 + window bounds 16 = 28 bytes
  -- Neu window chong nhau (hopping/sliding): state per account = [TODO: so cua so chong] x 28 bytes
  -- So account hoat dong dong thoi (uoc tinh): [TODO]
  -- Tong state: [TODO accounts] x [TODO bytes/account] = [TODO] MB
  -- Ket luan: co giu duoc trong RAM task khong? [TODO]
  -- Checkpoint state xuong durable storage vi: [TODO]

5. 💡 Gợi ý

5.1 Window type và event-time — hai câu hỏi phải trả lời

Câu 1 — hopping/sliding hay tumbling? Nhớ lại bài 03: một cơn gian lận kéo dài 5 phút bắt đầu lúc 12:03 có bị cửa sổ tumbling cố định [12:00,12:05)/[12:05,12:10) chẻ đôi và bỏ lỡ không? Yêu cầu "bất kỳ cửa sổ 5 phút nào" đòi loại window nào phủ mọi khoảng trượt? (Nếu một đợt vi phạm khiến nhiều cửa sổ chồng cùng vượt ngưỡng, bạn gặp vấn đề gì — và nên xử lý ở tầng window hay tầng alert?)

Câu 2 — event-time hay processing-time? Khi bạn reprocess 3 ngày lịch sử để kiểm tra, processing-time là thời điểm chạy lại (hôm nay), không phải lúc giao dịch xảy ra. Điều đó có nghĩa gì với cửa sổ tính toán? Với event-time, kết quả reprocess có khớp với kết quả production không? (Chốt ở §6.)

5.2 Watermark delay — trade-off latency vs completeness

Khi cửa sổ event-time đóng phụ thuộc vào watermark: window [T, T+5min) đóng khi watermark vượt T+5min. Watermark delay W nghĩa là bạn chờ thêm W giây so với event có timestamp lớn nhất để tính đến event trễ.

Câu hỏi định hướng: ATM/POS có delay dưới 5 giây; mobile app trễ 2–3 phút. Nếu bạn đặt watermark delay bằng 5 giây, mobile event đến sau 2 phút sẽ rơi vào tình trạng nào so với window đã đóng? Nếu đặt watermark delay bằng 3 phút để bắt mobile event, thì detection delay của pipeline tăng lên bao nhiêu phút so với SLA? Với bài toán fraud detection, yếu tố nào quan trọng hơn — bắt đủ 100% event hay phát cảnh báo sớm? (Chốt ở §6.)

Xử lý late event sau khi window đóng: allowed_lateness cho phép event trễ cập nhật window trong thêm một khoảng. Nếu late event đến sau allowed_lateness, không còn phương án nào khác ngoài side output. Bạn sẽ chọn ngưỡng allowed_lateness bao nhiêu để bắt mobile event mà không giữ state quá lâu?

5.3 Partition key — một câu hỏi trọng tâm

Kafka chỉ đảm bảo thứ tự trong một partition. Key quyết định partition cho mỗi record.

Đề bài yêu cầu phát hiện tổng vượt ngưỡng theo từng tài khoản. Nếu hai giao dịch của cùng một tài khoản rơi vào hai partition khác nhau (do chọn key sai), chúng bị xử lý bởi hai task khác nhau. Điều gì xảy ra với SUM(amount) mà mỗi task tính được, và nó ảnh hưởng ra sao tới việc phát hiện tài khoản vượt ngưỡng? Vậy partition key nên là gì để tránh điều đó? (Chốt ở §6.)

5.4 Exactly-once — hai thành phần, hai vai trò

Tự dựng kịch bản duplicate với at-least-once: trong vòng consume-process-produce (đọc event → xử lý → ghi alert vào sink → commit offset), consumer crash ở mốc nào thì lúc restart nó ghi lại alert lần hai vào fraud-alerts?

Kịch bản duplicate ở trên có nguồn gốc từ đâu — phía ghi của producer, hay từ việc consumer đã xử lý tới đâu (offset)? Exactly-once cần mấy tầng bảo vệ, và mỗi tầng chặn loại lỗi nào? Chỉ bật enable.idempotence=true (bảo vệ phía ghi) có đủ chặn alert trùng do consumer reprocess-after-crash không? (Chốt ở §6.)

6. ✅ Lời giải tham khảo

6.1 Sơ đồ pipeline

flowchart TB
  KA["Kafka: bank-transactions<br/>32 partition, key=account_id"]
  FP["Filter + Project<br/>loc tx hop le, chieu fields"]
  WA["Hopping Window 5min/hop 1min<br/>event-time, watermark 5s"]
  JT["Join KTable accounts<br/>non-windowed"]
  TH["Filter: total > alert_threshold"]
  KF["Kafka: fraud-alerts<br/>exactly-once"]
  SO["Kafka: fraud-late-events<br/>side output, allowed 3 min"]
  KTBL["KTable: accounts<br/>alert_threshold, account_name"]

  KA --> FP --> WA --> JT --> TH --> KF
  KTBL --> JT
  WA -->|"late events"| SO

6.2 Pseudocode đầy đủ

PIPELINE fraud_detection():

  -- ========================================================
  -- SOURCE: Kafka topic bank-transactions
  -- ========================================================
  -- Key = account_id (giu thu tu giao dich cung account trong partition)
  tx_stream <- kafka_source("bank-transactions")
                .assign_event_time(t -> t.tx_ts)           -- event-time tu terminal
                .assign_watermark(max_event_time - 5s)     -- watermark delay 5 giay

  -- ========================================================
  -- STEP 1: Filter + Project (stateless, no shuffle)
  -- ========================================================
  -- Chieu som: bo merchant_id, channel, tx_ts (khong dung trong aggregate)
  -- Giam payload truoc khi vao window operator (tiet kiem state + network)
  clean <- tx_stream
             .filter(t -> t.amount_vnd > 0)                -- loai tx amount am/zero
             .project(t -> (t.account_id, t.amount_vnd))   -- chi giu 2 field can thiet

  -- ========================================================
  -- STEP 2: Windowed Aggregate (stateful, shuffle on account_id)
  -- ========================================================
  -- HoppingWindow(size 5min, hop 1min): phu "bat ky 5 phut nao"
  -- Tumbling se che doi burst vat qua ranh gioi -> bo lo (bai 03)
  windowed <- clean
                .key_by(t -> t.account_id)                     -- moi account -> 1 task
                .window(HoppingWindow(5 min, hop 1 min, EventTime))
                .aggregate(
                    SUM(t.amount_vnd) as total_amount_vnd,
                    COUNT(*) as tx_count
                )
  -- Output per (account_id, window): (account_id, window_start, window_end,
  --   total_amount_vnd, tx_count)

  -- Late event: cho phep toi 3 phut sau window close (cap nhat window da fire)
  -- Sau 3 phut -> side output rieng, khong vao aggregate chinh
  windowed.late_events(allowed_lateness = 3 min)
            .sink_to_kafka("fraud-late-events", delivery = at_least_once)

  -- ========================================================
  -- STEP 3: Non-windowed Stream-Table Join
  -- ========================================================
  -- KTable tu topic accounts-changelog (compacted log, snapshot hien tai)
  -- Non-windowed join: an toan vi KTable co kich thuoc co dinh, khong phing state
  -- (Stream-stream join se can window -> state nhan doi tuyen tinh theo thoi gian)
  accounts_tbl <- kafka_source("accounts-changelog")
                    .as_table(key = "account_id")
  enriched <- windowed.join_table(accounts_tbl, key = "account_id")
  -- Output: + alert_threshold (nguong rieng), + account_name (cho alert message)

  -- ========================================================
  -- STEP 4: Filter nguong + Sink exactly-once
  -- ========================================================
  alerts <- enriched
              .filter(r -> r.total_amount_vnd > r.alert_threshold)

  alerts.sink_to_kafka("fraud-alerts",
                        key      = "account_id",            -- thu tu alert moi account
                        delivery = exactly_once,
                        mechanism = idempotent_producer     -- A: khong ghi duplicate du retry
                                  + kafka_transaction)      -- B: offset commit + write atomic
  -- Transaction bao gom: write(fraud-alerts) + commit(offset bank-transactions)
  -- Neu crash giua: ca hai roll back -> replay tu offset cu -> ket qua giong nhau

  -- ========================================================
  -- Uoc tinh state cua window operator
  -- ========================================================
  -- State per (account, window) = SUM 8 + COUNT 4 + window bounds 16 = 28 bytes
  -- Hopping size 5min / hop 1min => 5 cua so chong mo cung luc moi account
  -- => State per account = 5 x 28 = 140 bytes
  -- Active accounts dong thoi (uoc tinh VietBank): ~500.000
  -- Tong state: 500.000 x 140 bytes = 70 MB -- van vua RAM task (heap GB-level)
  -- Checkpoint state xuong durable storage (S3/HDFS) dinh ky de phuc hoi sau crash

6.3 Lập luận window type

Tại sao hopping/sliding, không phải tumbling?

Đây là bẫy dễ chọn sai. Yêu cầu là "tổng vượt ngưỡng trong bất kỳ cửa sổ 5 phút nào" — đúng tình huống bài 03 cảnh báo: một cơn gian lận kéo dài từ 12:03 tới 12:08 sẽ bị tumbling (cửa sổ cố định [12:00,12:05) và [12:05,12:10)) chẻ đôi — mỗi cửa sổ chỉ thấy một nửa tổng, không cửa nào vượt ngưỡng → bỏ lỡ. Chỉ hopping (size 5 phút, hop 1 phút) hoặc sliding mới phủ được "bất kỳ khoảng 5 phút nào".

Còn nỗi lo "một giao dịch xuất hiện ở nhiều cửa sổ chồng nhau → đếm chồng"? Đó không phải bug: mỗi cửa sổ là một câu hỏi độc lập "tổng trong [t, t+5min) là bao nhiêu", và một giao dịch thuộc nhiều cửa sổ chồng là đúng bản chất sliding. Cái cần xử lý là nhiều alert (mỗi cửa sổ vượt ngưỡng lại fire một lần) — giải bằng dedup: chỉ phát tối đa một alert cho mỗi (account, đợt vi phạm), ví dụ nén các alert trùng trong một khoảng cooldown. Đừng đổi sang tumbling chỉ để tránh alert trùng — vì tumbling bỏ lỡ đúng thứ cần bắt.

Tại sao event-time, không phải processing-time?

Processing-time là thời điểm event được xử lý bởi pipeline — phụ thuộc tải hệ thống, network delay, thời điểm chạy lại. Khi replay lịch sử để kiểm tra, tất cả event cũ xử lý vào "bây giờ" (processing-time hiện tại) → cửa sổ sai hoàn toàn. Event-time dùng tx_ts do terminal ghi lại → replay cho cùng kết quả → idempotent. Đây là lý do mọi production stream pipeline dùng event-time.

6.4 Lập luận watermark

Fraud detection ưu tiên phát hiện nhanh: kẻ gian có thể hoàn tất giao dịch trong vài phút. Watermark 5 giây bắt hầu hết ATM/POS (chiếm phần lớn giao dịch trong thực tế); mobile event trễ 2–3 phút rơi vào allowed_lateness (cập nhật window đã fire nếu còn trong 3 phút sau close) hoặc side output fraud-late-events để kiểm tra thủ công.

Trade-off được chấp nhận: một số giao dịch fraud qua mobile app có thể bị bỏ sót bởi aggregate chính. Tuy nhiên, nếu tăng watermark lên 3 phút để bắt đủ, toàn bộ pipeline chậm 3 phút → kẻ gian rút tiền xong trước khi cảnh báo đến.

6.5 Lập luận partition key

Kafka chỉ đảm bảo thứ tự và tính liên tục trong một partition. Window operator giữ state per key trên cùng một task — nếu cùng account_id về hai partition khác nhau, hai task khác nhau mỗi task chỉ thấy một phần giao dịch:

• Task A: nhìn thấy 3 giao dịch, tổng 8 triệu → không vượt ngưỡng 10 triệu → không alert
• Task B: nhìn thấy 4 giao dịch còn lại, tổng 6 triệu → không alert
• Thực tế 7 giao dịch tổng 14 triệu → phải alert nhưng không ai thấy

Key là account_id đảm bảo mọi giao dịch cùng tài khoản về cùng partition → cùng task → aggregate chính xác. Kafka Streams và Flink cũng dùng chính partition key này để route state, không cần shuffle thêm.

6.6 Lập luận exactly-once

Kịch bản duplicate với at-least-once:

1. Pipeline xử lý window [12:00, 12:05) cho account A — tổng 25 triệu, vượt ngưỡng.
2. Ghi alert vào fraud-alerts thành công.
3. Consumer crash trước khi commit offset về bank-transactions.
4. Restart từ offset cũ → replay window [12:00, 12:05) → ghi lại alert lần hai → alert trùng lặp.

Exactly-once kết hợp hai thành phần:

• Idempotent producer (enable.idempotence=true, từ Kafka 0.11): gắn producer ID + sequence number vào mỗi batch. Broker phát hiện và bỏ qua duplicate dù producer retry nhiều lần. Tuy nhiên, chỉ idempotent producer chưa đủ: sau crash consumer và restart, consumer vẫn replay → aggregate lại → ghi lần hai với sequence number mới → không phải duplicate theo góc nhìn của producer.
• Kafka transaction (transactional.id): bao gồm cả write(fraud-alerts) và commit(offset bank-transactions) trong cùng một transaction. Nếu crash giữa chừng, transaction abort → cả hai rollback → khi restart, consumer đọc từ offset cũ nhưng alert chưa commit → chỉ replay phần chưa hoàn thành → không double-count.

Overhead: idempotent producer gần 0% overhead (chỉ thêm header). Transaction giảm throughput khoảng 3% khi commit interval vài giây và batch đủ lớn — chấp nhận được cho bài toán fraud detection.

7. 🎓 Mở rộng

Ba tình huống để suy nghĩ thêm sau khi đã hoàn thành lời giải tham khảo:

M1 — Hot-partition skew: Một tài khoản tập đoàn thực hiện hàng nghìn giao dịch/giây (thanh toán batch tự động). Partition chứa account_id đó nhận gấp trăm lần lưu lượng trung bình → task xử lý partition đó trở thành straggler → cảnh báo cho tài khoản đó trễ hơn SLA 30 giây. Bạn có thể làm gì? (Gợi ý: key-salting account_id#bucket để trải sang nhiều task rồi gộp ở tầng aggregate thứ hai — KHÔNG sub-partition theo giờ, vì tách giao dịch cùng account làm vỡ aggregate 5 phút; hoặc tách luồng riêng cho tài khoản volume-cao.)

M2 — Cửa sổ linh hoạt theo tier tài khoản: Tài khoản PREMIUM yêu cầu sliding window 1 phút / step 10 giây (phát hiện nhanh hơn) thay vì hopping 5 phút/hop 1 phút. Bạn có tách thành hai pipeline song song không? Hay dùng một pipeline với dynamic window per account type? Trade-off của mỗi cách là gì về state size và độ phức tạp vận hành?

M3 — Replay lịch sử an toàn: Sau deploy, team muốn replay 3 ngày giao dịch cũ (còn trong Kafka, retention 7 ngày) để kiểm tra pipeline mới có phát hiện được fraud đã xảy ra không. Bạn cần thay đổi gì để replay không ghi alert thật vào fraud-alerts production? Cần cấu hình gì để watermark xử lý đúng dữ liệu lịch sử (tất cả event đến cùng lúc nhưng tx_ts rải suốt 3 ngày)?

8. ✨ Điều bạn vừa làm được

• Lập luận window type bằng ngữ nghĩa (bài 03): "bất kỳ cửa sổ 5 phút nào" đòi hopping/sliding — tumbling chẻ đôi burst vắt qua ranh giới nên bỏ lỡ. Việc một giao dịch rơi vào nhiều cửa sổ chồng là đúng bản chất sliding, xử lý bằng dedup alert chứ không đổi window. Ngữ nghĩa yêu cầu quyết định loại window, không phải thói quen.

• Chọn watermark dựa trên SLA và đặc tính nguồn (bài 03): cân bằng giữa watermark delay và detection latency theo channel cụ thể (ATM dưới 5s vs mobile 2–3 phút); chấp nhận miss mobile event qua side output thay vì tăng watermark làm chậm toàn pipeline. Đây là quyết định latency-vs-completeness thực tế, không phải chọn con số ngẫu nhiên.

• Stream-table join cho enrichment (bài 04): hiểu KTable là changelog topic dạng compacted log → non-windowed join an toàn, state giới hạn bởi số tài khoản (không phình theo thời gian). Nếu dùng stream-stream join thay KTable phải thêm window → state tăng tuyến tính vô hạn.

• Phân tích exactly-once theo kịch bản cụ thể (bài 04): không chỉ bật enable.idempotence=true rồi tin là xong, mà hiểu tại sao cần thêm Kafka transaction: idempotent producer chặn duplicate ở tầng producer, nhưng không ngăn consumer replay và aggregate lại sau crash. Transaction biến offset commit + alert write thành một atomic unit — đây mới là exactly-once thật sự end-to-end.

Bài tiếp theo: Tổng kết module — cheat sheet & self-assessment