Flink 3 — Event time, Watermark & Windows

Vì sao "thời gian" là bài toán khó nhất của stream processing

Trong xử lý luồng, gần như mọi phép tổng hợp có ý nghĩa đều gắn với thời gian: "đếm giao dịch mỗi 5 phút", "tổng số tiền chuyển trong 1 giờ trượt", "phiên đăng nhập của khách hàng". Nhưng "thời gian" trong hệ phân tán không đơn giản như đồng hồ treo tường. Một sự kiện giao dịch có ba mốc thời gian rất khác nhau, và chọn nhầm mốc sẽ cho kết quả sai lệch mà không hề báo lỗi.

Đây là bài lõi và khó nhất của series Flink. Nắm được event time, watermark và window là nắm được phần lớn giá trị của Flink so với công cụ batch. Bài giả định bạn đã hiểu mô hình dataflow (nguồn, toán tử, keyed stream, song song hoá).

Ba khái niệm thời gian

Ví dụ cụ thể ở ngân hàng: một khách hàng quẹt thẻ POS lúc 10:00:00 (event time). Nhưng máy POS mất mạng, tin nhắn được gom lại và đẩy về Kafka lúc 10:02:30; Flink đọc và tổng hợp lúc 10:02:31 (processing time). Nếu ta đếm giao dịch "cửa sổ 10:00–10:01" theo processing time, giao dịch này sẽ rơi nhầm vào cửa sổ 10:02 — sai bản chất.

Vì sao event time cho kết quả đúng đắn & nhất quán

Điểm mấu chốt: event time cho kết quả xác định (deterministic). Cùng một tập dữ liệu, dù bạn chạy live hôm nay hay replay lại từ Kafka tuần sau, dù cluster nhanh hay chậm, giao dịch quẹt thẻ lúc 10:00 luôn thuộc về cửa sổ 10:00. Kết quả không đổi.

Ngược lại, processing time cho kết quả phụ thuộc thời điểm và tốc độ chạy: replay lại sẽ ra số khác vì đồng hồ máy đã khác. Với nghiệp vụ ngân hàng cần đối soát, kiểm toán và tái tạo lại báo cáo, tính bất biến của event time là điều kiện bắt buộc. Processing time chỉ nên dùng khi bạn thực sự cần độ trễ cực thấp và không quan tâm chính xác về mặt thời điểm (ví dụ: đếm số message thô đang chảy qua để giám sát tải).

Dữ liệu đến trễ và không đúng thứ tự

Nếu event time hoàn hảo như vậy, tại sao không dùng mặc định? Vì nó kéo theo một vấn đề nan giải: out-of-order (không đúng thứ tự) và late data (đến trễ).

Trong thực tế, các sự kiện không đến Flink theo đúng thứ tự event time. Máy POS A gửi sự kiện 10:00:05 xong mới có mạng lại và gửi sự kiện 10:00:01 bị kẹt trước đó. Nhiều partition Kafka, nhiều nguồn song song, mạng chập chờn — tất cả khiến dòng sự kiện đến Flink bị xáo trộn theo event time.

Câu hỏi cốt lõi: khi nào Flink được phép "chốt" cửa sổ 10:00–10:01 và phát ra kết quả? Nếu chốt ngay khi thấy sự kiện đầu tiên có event time ≥ 10:01, ta có thể bỏ sót các sự kiện 10:00:xx đến trễ. Nếu chờ mãi không chốt, kết quả không bao giờ được phát ra. Đây là bài toán đánh đổi giữa độ trễ (latency — chốt sớm) và độ hoàn chỉnh (completeness — chờ đủ dữ liệu). Watermark chính là cơ chế Flink dùng để giải bài toán này.

Watermark — mốc "đã thấy hết sự kiện tới thời điểm T"

Watermark là một bản ghi đặc biệt luồn trong dòng dữ liệu, mang giá trị timestamp W(t). Ý nghĩa: "Tôi khẳng định sẽ không còn (hoặc gần như không còn) sự kiện nào có event time ≤ t đến sau điểm này nữa." Nói cách khác, watermark là lời tuyên bố về mức độ tiến triển của event time trong dòng.

Khi một toán tử window nhận watermark W(t), nó biết mọi cửa sổ có thời điểm kết thúc ≤ t đã "đủ dữ liệu" và có thể được kích hoạt (trigger) để tính toán rồi phát ra kết quả. Watermark chính là cái đồng hồ event-time của Flink — nó tiến lên theo dữ liệu, không theo đồng hồ máy.

Watermark generator: bounded out-of-orderness

Làm sao Flink biết đặt watermark ở đâu? Chiến lược phổ biến nhất là bounded out-of-orderness (giả định độ trễ có giới hạn): ta cho rằng sự kiện có thể đến trễ tối đa một khoảng B so với sự kiện muộn nhất đã thấy. Watermark được tính bằng:

W = max(event_time đã thấy) - B

Ví dụ minh hoạ (Flink DataStream API, Java — không phải SQL chạy được trong sandbox):

// Minh hoạ: khai báo watermark strategy cho luồng giao dịch
WatermarkStrategy<Transaction> strategy = WatermarkStrategy
    .<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(30))
    .withTimestampAssigner((tx, ts) -> tx.getEventTime())   // lấy event time từ bản ghi
    .withIdleness(Duration.ofMinutes(1));                    // xử lý partition im lặng

DataStream<Transaction> withWm = source.assignTimestampsAndWatermarks(strategy);

Ở đây B = 30 giây. Nếu Flink đã thấy sự kiện muộn nhất là 10:05:00, watermark hiện tại là 10:04:30 — nghĩa là "coi như đã đủ dữ liệu tới 10:04:30". Cửa sổ kết thúc ≤ 10:04:30 sẽ được chốt.

Chọn B là nghệ thuật đánh đổi: B lớn → chờ lâu, độ trễ cao nhưng ít bỏ sót; B nhỏ → kết quả nhanh nhưng nhiều sự kiện bị coi là "trễ". Với giao dịch POS ngân hàng có nguồn mạng chập chờn, B thường đặt vài chục giây đến vài phút. Ngoài forBoundedOutOfOrderness, còn có forMonotonousTimestamps (khi nguồn đảm bảo tăng đơn điệu, B=0) và generator tự viết.

withIdleness rất quan trọng: nếu một partition Kafka không có dữ liệu, nó không phát watermark, kéo tụt watermark toàn cục và làm cửa sổ không bao giờ chốt. withIdleness cho phép bỏ qua partition im lặng đó.

Watermark lan truyền qua đồ thị

Watermark không đứng yên — nó chảy qua đồ thị toán tử cùng dữ liệu. Quy tắc lan truyền then chốt: một toán tử có nhiều đầu vào lấy watermark bằng GIÁ TRỊ NHỎ NHẤT (min) trong các đầu vào của nó.

Điều này logic: nếu toán tử gộp (ví dụ sau keyBy hoặc union) nhận từ nguồn A watermark 10:05 nhưng nguồn B mới tới 10:02, thì toán tử chỉ dám khẳng định "đã thấy hết tới 10:02" — vì B có thể còn sự kiện cũ chưa gửi. Đây là lý do một nguồn/partition chậm (straggler) hoặc im lặng có thể "ghìm" watermark toàn hệ thống, khiến mọi cửa sổ downstream bị treo.

Đây là điểm khác biệt lớn so với watermark trong Spark Structured Streaming: Flink cho watermark là công dân hạng nhất, lan truyền theo record-by-record qua toàn đồ thị, cho phép điều khiển thời gian tinh vi hơn ở cấp toán tử.

Windows — chia dòng vô hạn thành các khối hữu hạn

Dòng dữ liệu là vô hạn, nhưng phép tổng hợp cần biên. Window (cửa sổ) là cơ chế gom các sự kiện trong một khoảng event-time nhất định lại để tính toán. Flink có bốn loại window chính.

Tumbling — cửa sổ cố định, không chồng lấn

Chia trục thời gian thành các đoạn liên tiếp, bằng nhau, không đè lên nhau. Mỗi sự kiện thuộc đúng một cửa sổ. Ví dụ tumbling 5 phút: [10:00–10:05), [10:05–10:10)... Dùng khi cần báo cáo định kỳ không trùng lặp: "đếm giao dịch mỗi 5 phút".

Sliding — cửa sổ trượt, chồng lấn

Có hai tham số: kích thước (size) và bước trượt (slide). Nếu size > slide, các cửa sổ đè lên nhau và mỗi sự kiện thuộc nhiều cửa sổ. Ví dụ size 10 phút, slide 1 phút: mỗi phút phát ra một tổng của 10 phút gần nhất. Dùng cho chỉ số dạng "trung bình trượt", phát hiện đột biến mượt hơn tumbling.

Session — cửa sổ theo khoảng nghỉ hoạt động

Không cố định biên; cửa sổ được xác định bằng gap (khoảng lặng). Nếu giữa hai sự kiện của cùng một key cách nhau quá gap, chúng thuộc hai session khác nhau. Ví dụ gap 30 phút: một chuỗi thao tác của khách trên app mobile banking được gom thành một phiên; ngừng quá 30 phút thì mở phiên mới. Rất tự nhiên cho phân tích hành vi người dùng.

Global — một cửa sổ cho tất cả

Gom mọi sự kiện của một key vào một cửa sổ duy nhất, không bao giờ tự chốt — bạn phải tự cấp trigger (ví dụ đếm đủ 100 phần tử thì kích hoạt). Dùng cho các mẫu count-based tuỳ biến.

Window assigner + trigger + evictor

Mỗi window thực chất là tổ hợp ba thành phần, và hiểu rạch ròi giúp bạn tuỳ biến sâu:

• Window assigner: quyết định mỗi sự kiện thuộc (những) cửa sổ nào. Ví dụ TumblingEventTimeWindows.of(Time.minutes(5)).
• Trigger: quyết định khi nào cửa sổ được tính và phát kết quả. Trigger mặc định của event-time window là EventTimeTrigger — kích hoạt khi watermark vượt qua thời điểm kết thúc cửa sổ. Ta có thể viết trigger để phát kết quả sớm (early firing) trước khi watermark tới, phục vụ dashboard gần realtime.
• Evictor (tuỳ chọn): loại bỏ một số phần tử khỏi cửa sổ trước/sau khi hàm window chạy (ví dụ giữ lại N phần tử cuối). Ít dùng, chỉ khi cần logic đặc biệt.

// Minh hoạ: tumbling event-time window 5 phút, đếm giao dịch theo tài khoản
withWm
    .keyBy(Transaction::getAccountId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .allowedLateness(Time.minutes(1))
    .sideOutputLateData(lateTag)
    .aggregate(new CountAgg());

Xử lý sự kiện đến trễ: allowed lateness & side output

Watermark đã cho ta điểm chốt cửa sổ, nhưng vẫn còn sự kiện đến sau watermark. Flink có hai lớp phòng thủ.

Allowed lateness: sau khi cửa sổ đã chốt (watermark vượt qua điểm kết thúc), Flink vẫn giữ trạng thái cửa sổ thêm một khoảng allowedLateness. Nếu trong khoảng đó có sự kiện trễ tới đúng cửa sổ, Flink kích hoạt lại cửa sổ và phát ra kết quả cập nhật (một bản ghi mới ghi đè bản trước). Điều này cho phép sửa số liệu khi dữ liệu trễ về, đổi lại là downstream phải xử lý được kết quả cập nhật.

Side output cho dữ liệu quá trễ: khi sự kiện đến sau cả watermark + allowedLateness, nó bị coi là "quá trễ" (too late) và mặc định bị loại bỏ âm thầm — nguy hiểm vì mất dữ liệu không dấu vết. Giải pháp: khai báo OutputTag và sideOutputLateData(lateTag) để định tuyến các bản ghi quá trễ ra một luồng phụ (side output), rồi ghi vào bảng "late events" để đối soát, cảnh báo hoặc xử lý riêng.

Window function — cách tính bên trong cửa sổ

Sau khi biết một cửa sổ gồm những sự kiện nào, bạn cần một hàm để tính kết quả. Ba loại chính, đánh đổi giữa hiệu năng và khả năng biểu đạt:

Mẹo thực chiến: kết hợp AggregateFunction (tính gộp tăng dần, tiết kiệm bộ nhớ) với ProcessWindowFunction (chỉ để lấy metadata cửa sổ khi phát kết quả). Flink cho phép truyền cả hai vào cùng lời gọi .aggregate(aggFn, processFn) — vừa nhẹ RAM vừa có ngữ cảnh cửa sổ. Trạng thái mà window nắm giữ được quản lý qua state backend, chủ đề của bài quản lý trạng thái.

Use case thực tế: phát hiện giao dịch bất thường dù đến trễ

Bối cảnh. Đội giám sát gian lận NCB cần cảnh báo khi một tài khoản phát sinh > 20 giao dịch trong bất kỳ cửa sổ 5 phút nào — dấu hiệu điển hình của tấn công dò thẻ (card testing) hoặc bot rút tiền. Nguồn giao dịch POS về qua Kafka, nhưng một số máy POS ở vùng sóng yếu gửi trễ tới 20–40 giây, và thứ tự bị xáo trộn.

Nếu dùng processing time (như cách batch mỗi 5 phút): các giao dịch tấn công đến trễ sẽ rơi vào cửa sổ sai, đếm không đủ 20 trong bất kỳ cửa sổ nào → bỏ lọt cảnh báo. Đúng lúc cần thì hệ thống im lặng.

Thiết kế bằng Flink event-time:

1. Watermark: forBoundedOutOfOrderness(Duration.ofSeconds(30)), lấy event time từ trường posTimestamp của bản ghi. Bật withIdleness(1 phút) để các POS im lặng ban đêm không ghìm watermark.
2. Key & window: keyBy(accountId), tumbling event-time 5 phút. Mỗi tài khoản có chuỗi cửa sổ 5 phút độc lập.
3. Aggregate: AggregateFunction đếm số giao dịch tăng dần (nhẹ RAM, không buffer cả cửa sổ).
4. Allowed lateness 1 phút: giao dịch trễ trong vòng 1 phút sau khi cửa sổ chốt vẫn được cộng vào và cảnh báo được tính lại — không bỏ sót do trễ nhẹ.
5. Side output: giao dịch tới sau watermark + 1 phút được đẩy sang luồng too-late-tx, ghi vào bảng đối soát để đội vận hành rà thủ công.
6. Trigger cảnh báo: khi kết quả cửa sổ vượt ngưỡng 20, phát sự kiện FraudAlert(accountId, windowStart, count) sang topic cảnh báo.

Kết quả định lượng (minh hoạ tình huống). Trong một ca thử nghiệm với 3 tài khoản bị tấn công, mỗi tài khoản có 32–45 giao dịch/5 phút nhưng ~15% giao dịch đến trễ 20–35 giây: phương án processing-time chỉ bắt được 1/3 tài khoản; phương án event-time + watermark 30s + allowed lateness 1 phút bắt đủ 3/3, độ trễ cảnh báo trung bình ~35 giây sau khi cửa sổ kết thúc. Con số minh hoạ để làm rõ tác động của lựa chọn thời gian, không phải benchmark chính thức.

Về mặt tư duy nghiệp vụ, logic "đếm giao dịch theo cửa sổ thời gian" này rất gần với các luật trong giám sát giao dịch AML — nơi tính đúng đắn của mốc thời gian ảnh hưởng trực tiếp tới việc có bỏ lọt hành vi đáng ngờ hay không.

Ghi nhớ

• Ba loại thời gian: event time (lúc xảy ra thực tế — bất biến, cho kết quả xác định khi replay), processing time (lúc xử lý — rẻ, độ trễ thấp, không tái tạo được), ingestion time (lúc vào Flink). Nghiệp vụ cần đối soát/kiểm toán → luôn dùng event time.
• Vấn đề gốc: event time kéo theo dữ liệu đến trễ và không đúng thứ tự (out-of-order); phải cân bằng độ trễ (chốt sớm) với độ hoàn chỉnh (chờ đủ).
• Watermark W(t) = tuyên bố "đã thấy hết sự kiện tới thời điểm t". Là đồng hồ event-time của Flink, kích hoạt cửa sổ khi vượt qua điểm kết thúc.
• Bounded out-of-orderness: W = max(event_time) - B. B lớn → chậm mà đầy đủ; B nhỏ → nhanh mà dễ mất. Bật withIdleness để partition im lặng không ghìm watermark.
• Lan truyền watermark = min các đầu vào: một nguồn chậm/straggler ghìm toàn hệ thống.
• Bốn loại window: tumbling (cố định, không chồng), sliding (trượt, chồng lấn), session (theo gap hoạt động), global (tự cấp trigger). Mỗi window = assigner + trigger + evictor(tuỳ chọn).
• Xử lý trễ hai lớp: allowed lateness (chốt rồi vẫn cập nhật lại kết quả) và side output (định tuyến dữ liệu quá trễ ra luồng riêng, tránh mất dữ liệu âm thầm).
• Window function: ReduceFunction/AggregateFunction tính tăng dần (nhẹ RAM); ProcessWindowFunction buffer cả cửa sổ nhưng có đủ metadata. Kết hợp aggregate + process để vừa nhẹ vừa đủ ngữ cảnh.