Windowing — cắt dòng vô tận thành cửa sổ hữu hạn để tổng hợp

TL;DR: Stream processing xử lý dòng sự kiện vô tận — nhưng muốn tổng hợp (đếm, tính trung bình, phát hiện bất thường), ta cần cắt dòng vô tận thành cửa sổ hữu hạn gọi là window. Có bốn loại cơ bản: tumbling (không chồng), hopping (chồng cố định), sliding (cập nhật liên tục), session (tự co giãn theo hành vi). Điểm khoá là phân biệt event-time (timestamp do producer gắn, thời điểm sự kiện thật xảy ra) và processing-time (thời điểm server nhận được) — chỉ event-time mới cho phép reprocess lịch sử ra cùng kết quả. Watermark(t) khai báo "sự kiện với timestamp nhỏ hơn hoặc bằng t coi như đã đến đủ" để window có thể đóng lại — cân bằng giữa latency và completeness.

Hệ thống giám sát gian lận của một ngân hàng phải phát hiện nếu một thẻ quẹt hơn 5 lần trong vòng 5 giây — không phải "5 giây kể từ khi server nhận được tin", mà là "5 giây theo đồng hồ thật của thiết bị quẹt". Khi mạng chập chờn, sự kiện tới trễ 30 giây; khi reprocess log lịch sử, sự kiện "tới" theo thời gian thực của cả năm trước chỉ trong vài phút. Nếu dùng thời gian server (processing-time), reprocess cho ra kết quả sai hoàn toàn. Nếu dùng event-time mà không có watermark, window không biết bao giờ đã đủ sự kiện để đóng lại — state tích luỹ vô hạn.

Bài này giải thích cách window cắt dòng vô tận thành tính toán hữu hạn, vì sao event-time là lựa chọn mặc định khi cần tính lại được, và watermark hoạt động ra sao để giải quyết bài toán "chờ bao lâu là đủ".

1. Analogy — trạm đếm xe qua cầu mỗi 5 phút

Tưởng tượng bạn đứng ở trạm kiểm soát đầu cầu, đếm số xe qua trong mỗi khoảng 5 phút để báo cáo lưu lượng. Xe đến liên tục — không bao giờ có lúc "hết xe". Bạn không thể chờ đến khi "tất cả xe đã qua" mới đếm, vì điều đó là mãi mãi.

Thay vào đó, bạn bấm đồng hồ: mỗi 5 phút, đọc số đếm hiện tại, ghi xuống, reset về 0, đếm tiếp. Đây chính là tumbling window: cắt dòng vô tận thành những ô 5 phút không chồng nhau.

Nhưng có một chi tiết tinh tế: bạn đếm theo đồng hồ của xe (thời điểm xe thực sự qua cầu) hay đồng hồ của trạm (thời điểm trạm camera ghi nhận)? Nếu sương mù làm camera delay 30 giây, hai đồng hồ lệch nhau — và nếu sau đó bạn xem lại băng để kiểm tra, kết quả sẽ rất khác.

2. Vì sao dòng vô tận không có 'kết quả cuối'?

Gọi count() trên một list hữu hạn — có kết quả. Gọi count() trên một stream vô tận — không bao giờ trả về, vì stream không bao giờ kết thúc.

# Bounded input:
count([e1, e2, e3, e4, e5])  --> 5  (biet diem cuoi)

# Unbounded stream:
count(stream)                --> ??? (khong bao gio xong)

Giải pháp là scope phép tính vào window: thay vì "tổng hợp toàn bộ stream", ta tổng hợp "trong mỗi window 5 phút". Window biến bài toán vô tận thành loạt bài toán hữu hạn nối tiếp nhau.

flowchart LR
  STREAM["Stream vo tan<br/>(e1 e2 e3 ...)"] --> W1["Window 0-5 phut<br/>ket qua: 120"]
  STREAM --> W2["Window 5-10 phut<br/>ket qua: 98"]
  STREAM --> W3["Window 10-15 phut<br/>ket qua: 135"]
  W1 & W2 & W3 --> OUT["Chuoi ket qua<br/>tiep tuc xu ly"]

Điểm quan trọng: window không làm dừng stream. Stream vẫn chảy liên tục; window chỉ xác định phạm vi của từng phép tính. Khi window 0–5 phút đóng và cho ra kết quả, sự kiện trong khoảng 5–10 phút đã tích luỹ trong window tiếp theo.

3. Bốn loại window

Bốn loại window phân biệt nhau ở cách phân bổ sự kiện và cách các window chồng lên nhau.

3.1 Tumbling window — không chồng, mỗi sự kiện thuộc đúng 1 window

Chia timeline thành các khoảng đều nhau, không chồng nhau. Mỗi sự kiện thuộc đúng 1 window.

|--- W1 (0-5 phut) ---||--- W2 (5-10 phut) ---||--- W3 (10-15 phut) ---|
 e1 e2 e3              e4 e5                    e6 e7 e8

Dùng khi: tính tổng hoặc trung bình theo chu kỳ cố định (doanh thu theo giờ, số request theo phút).

3.2 Hopping window — chồng cố định, mỗi sự kiện có thể ở nhiều window

Kích thước window cố định nhưng hop nhỏ hơn kích thước — window chồng nhau. Tham số: size và hop.

size=10 phut, hop=5 phut:
|------ W1 (0-10) ------|
          |------ W2 (5-15) ------|
                    |------ W3 (10-20) ------|

Dùng khi: "cứ 5 phút báo cáo số request trong 10 phút gần nhất".

3.3 Sliding window — cập nhật liên tục

Về bản chất là hopping với hop rất nhỏ: mỗi sự kiện mới vào kích hoạt tính lại toàn bộ window. Lưu ý thuật ngữ theo framework: Flink gọi window (size, slide) là "sliding" — đúng bằng cái bài này gọi hopping; còn "sliding window" của Kafka Streams (KIP-450) không có tham số slide mà chỉ có timeDifference (độ rộng), tính lại mỗi khi có event mới. Ý tưởng chung: window trượt liên tục theo thời gian.

Dùng khi: phát hiện bất thường liên tục (rate limit: "5 request trong bất kỳ khoảng 10 giây nào").

3.4 Session window — tự co giãn theo hành vi người dùng

Không có kích thước cố định — window kéo dài khi sự kiện đến liên tiếp, đóng lại khi im lặng quá session_gap. Mỗi người dùng có session window riêng.

session_gap = 30 phut:
User A:  |-- session 1 --|   (im lang > 30 phut)   |-- session 2 --|
           e1 e2 e3                                   e4 e5

User B:  |--- session (dai hon) ---|
           e1  e2      e3  e4

Dùng khi: phân tích phiên làm việc ("mỗi lần user dùng app, tổng thời gian là bao nhiêu?").

Đáp án: Sliding window (hoặc hopping với hop nhỏ). Tumbling 5 phút bỏ lỡ trường hợp load cao bắt đầu ở phút 4 và kéo đến phút 7 — nằm ở hai window khác nhau, trung bình mỗi window thấp dù thực tế có 3 phút liên tiếp vượt 80%. Session window không phù hợp vì không dựa trên khoảng im lặng. Sliding window kiểm tra liên tục "bất kỳ 5 phút nào".

4. Event-time hay processing-time — khi nào dùng cái nào?

Processing-time là thời điểm hệ thống xử lý nhận được sự kiện — đơn giản nhất để implement vì không cần timestamp từ producer.

Event-time là thời điểm sự kiện thực sự xảy ra ở nguồn — producer gắn timestamp này vào sự kiện trước khi gửi đi.

Đáp án:

• Event-time: 14:02 (khi người dùng thực sự click)
• Processing-time: 14:05:03 (khi server nhận được)
• Processing-time window: rơi vào window 14:05–14:10 (đến sau khi window 14:00–14:05 đã đóng)
• Event-time window: thuộc window 14:00–14:05 (đúng ý nghĩa nghiệp vụ — click xảy ra trong khoảng đó)

Sự lệch nhau giữa event-time và processing-time gọi là event-time skew. Theo số liệu thực tế: mobile offline flush về trễ 2–5 phút là bình thường; IoT sensor batch gửi có thể trễ hàng giờ. Fraud detection watermark 5 giây (ưu tiên latency); financial reconciliation allowed lateness 1 giờ (ưu tiên completeness) — watermark định nghĩa ở mục 5, allowed lateness ở mục 6.

4.1 Tại sao event-time cho phép reprocess mà processing-time không?

PROCESSING-TIME (sai khi reprocess):
  Reprocess luc 3:00 AM ngay hom sau
  -> "processing-time" cua moi su kien = 3:00 AM hom sau
  -> tat ca su kien roi vao window 3:00-3:05 AM -> vo nghia nghiep vu

EVENT-TIME (dung khi reprocess):
  Reprocess luc 3:00 AM ngay hom sau
  -> "event-time" van giu nguyen timestamp goc (9:02, 9:03, 14:02...)
  -> window 9:00-9:05, 14:00-14:05... van dung nhu khi chay live

Xử lý theo event-time có nghĩa là "window phân loại sự kiện theo khi nào nó xảy ra, không phải khi nào ta biết đến nó". Đây là lý do Dataflow Model (VLDB 2015) coi event-time là nền tảng của stream processing đúng đắn — processing-time window và batch reprocessing về cơ bản không tương thích nhau.

5. Watermark — lower-bound trên event-time

Nếu window theo event-time, hệ thống cần biết khi nào window có thể đóng lại. Không thể chờ mãi mãi (state vô hạn), nhưng đóng quá sớm thì bỏ sự kiện trễ.

Watermark là câu trả lời: một tín hiệu đặc biệt trong stream, khai báo rằng event-time đã đạt đến một mốc nhất định. Định nghĩa chính thức từ Apache Flink docs:

"Watermark(t) declares event time has reached t — no more elements with timestamp t' ≤ t."

Khi watermark vượt qua điểm cuối của một window, operator biết window đó đã "đóng" và có thể tổng hợp kết quả.

flowchart LR
  STREAM["e(9:01) e(9:03) WM(9:04) e(9:02-late) WM(9:07)"] --> PROC["Window operator"]
  PROC -->|"nhan WM(9:04): chua du, window 9:00-9:05 con mo"| OPEN["Window 9:00-9:05<br/>(dang tich luy)"]
  PROC -->|"nhan WM(9:07): watermark vuot 9:05 -> dong"| RESULT["Ket qua window 9:00-9:05"]

5.1 Trade-off latency vs completeness

Watermark được tính theo heuristic thường dùng: max_event_time_seen - expected_max_delay. Nó là ước tính, không phải đảm bảo tuyệt đối.

Hai thái cực:

• Watermark delay 30 giây: window đóng sớm, latency thấp, kết quả nhanh. Nhưng dưới 1% event trễ hơn 30 giây sẽ bị coi là late event (đến sau window đóng).
• Watermark delay 5 phút: chờ đủ lâu để bắt hầu hết event trễ, completeness cao. Nhưng mọi window phải chờ thêm 5 phút sau khi "đáng ra" đóng — latency toàn hệ thống tăng 5 phút.

Theo số liệu thực tế từ brief: Azure Stream Analytics khuyến nghị IoT bắt đầu với watermark 5 phút; fraud detection có thể dùng 5 giây (ưu tiên latency). Không có con số đúng tuyệt đối — phụ thuộc vào SLA của nguồn và yêu cầu nghiệp vụ.

6. Late event — allowed lateness và side output

Watermark delay không bao giờ "bắt được hết" — một số sự kiện vẫn đến sau khi window đã đóng (late event). Có ba cách xử lý:

Option 1 — Loại bỏ (discard): đơn giản nhất, nhưng mất dữ liệu.

Option 2 — Allowed lateness: cho phép window ở lại trạng thái "mở thêm" một khoảng thời gian sau khi watermark qua điểm cuối. Event trễ trong khoảng này vẫn được tính vào window, và window phát lại kết quả cập nhật (late firing). Ví dụ financial reconciliation: allowed lateness 1 giờ — mọi giao dịch đến trong vòng 1 giờ sau khi window đóng vẫn được cộng vào.

Option 3 — Side output: event đến quá muộn (vượt allowed lateness) được route sang một stream riêng để xử lý thủ công hoặc theo logic đặc biệt — không bị mất hoàn toàn, nhưng không lẫn vào kết quả chính.

Timeline:
  Window [9:00, 9:05)   WM(9:09)        WM(9:14)
  |<-- dong khi WM(9:09) den -->|<-- allowed_lateness 5 phut -->|

  e(9:03) den luc 9:04  -> TRONG window, tinh binh thuong
  e(9:02) den luc 9:10  -> LATE nhung trong allowed_lateness -> cap nhat ket qua window
  e(9:01) den luc 9:16  -> VUOT allowed_lateness -> side output stream

Trong Apache Flink, cả ba cơ chế đều có API: .allowedLateness(Duration) và .sideOutputLateData(tag).

7. Pitfall tổng hợp

❌ Nhầm 1 — Dùng processing-time thay event-time:

❌ SAI:
  window = tumbling(5 phut, PROCESSING-TIME)
  reprocess log thang truoc luc 3 AM
  -> tat ca su kien "xay ra" luc 3 AM theo processing-time
  -> window 3:00-3:05 AM co 50 trieu event -> vo nghia nghiep vu

✅ DUNG:
  window = tumbling(5 phut, EVENT-TIME)
  reprocess log thang truoc -> event-time giu nguyen timestamp goc
  -> window 9:00-9:05 thang truoc van dung su kien dung khoang do

Mọi hệ thống stream production-grade (Flink, Kafka Streams, Google Dataflow) mặc định dùng event-time.

❌ Nhầm 2 — Watermark delay = 0:

❌ SAI: watermark delay = 0
  -> window dong ngay khi max event-time seen = diem cuoi window
  -> bat ky event nao den sau 1ms la "late" -> mat
  -> IoT sensor delay 30s binh thuong: mat 100% event tu sensor do

✅ DUNG: watermark delay >= P99 latency cua nguon
  IoT delay 30s binh thuong -> dat delay >= 30s
  + allowed_lateness cho long duoi it gap

❌ Nhầm 3 — Không watermark → window không đóng:

❌ SAI: event-time window nhung KHONG co watermark strategy
  -> operator khong biet "event nao la cuoi" cho moi window
  -> window khong dong, state tich luy vo han
  -> OutOfMemoryError hoac pipeline khong cho ra ket qua

✅ DUNG: luon khai bao watermark strategy
  - Event-time monotonic (sen sor thu tu dam bao) -> bounded disorder = 0
  - Event-time khong thu tu -> dat delay theo P99 latency cua nguon

8. 📚 Deep Dive

9. Liên hệ các bài khác

• Bài 02 — Log-based broker: Kafka giữ log với offset là nền tảng mà windowing xây trên đó; Kafka Streams triển khai tumbling/hopping/session window với event-time trực tiếp trên Kafka topic.
• Bài 04 — Stream join và exactly-once: stream-stream join bắt buộc có window — nếu không, state của hai stream tích luỹ vô hạn; bốn loại window ở bài này áp dụng trực tiếp khi thiết kế join.
• Bài 05 — Thiết kế stream pipeline: lab tổng hợp nơi bạn phải chọn loại window và watermark delay phù hợp với yêu cầu bài toán cụ thể.
• Bài 01 (Module 1) — Tư duy batch: mặt đối lập — batch xử lý input bounded không cần window vì có điểm kết thúc rõ ràng; so sánh giúp thấy rõ vì sao unbounded input làm nảy sinh nhu cầu windowing.

10. Tóm tắt

• Window cắt stream unbounded thành từng đoạn hữu hạn để tổng hợp — vì stream không có "điểm kết thúc" để tính kết quả cuối.
• Bốn loại: tumbling (không chồng, 1 sự kiện vào đúng 1 window), hopping (chồng cố định), sliding (cập nhật liên tục), session (tự co giãn theo session_gap).
• Event-time (timestamp producer) vs processing-time (đồng hồ server): event-time mới cho phép reprocess lịch sử ra cùng kết quả; processing-time không thể reprocess đúng.
• Watermark(t): lower-bound trên event-time — khai báo "event với timestamp ≤ t đã đến đủ". Window đóng khi watermark vượt điểm cuối. Watermark delay = trade-off latency vs completeness.
• Late event: sự kiện đến sau khi window đóng. Xử lý bằng: discard / allowed lateness (cập nhật kết quả trong grace period) / side output (route sang stream riêng).
• Pitfall lớn nhất: processing-time thay event-time → reprocess sai; watermark delay = 0 → mất event trễ; không watermark → window không đóng, state vô hạn.

11. Tự kiểm tra

Bài tiếp theo: Stream join và exactly-once