Thời gian trong Apache Flink: watermark, phân vùng rảnh, state TTL và replay

Trong stream processing, phần lớn lỗi khó tìm không nằm ở logic tính toán mà ở thời gian: sự kiện đến muộn, một phân vùng im lặng, state hết hạn theo đồng hồ nào, và một job chạy lại có ra cùng kết quả hay không.

Bài này ghi lại những gì mình học được về thời gian trong Apache Flink qua một pipeline phân tích sự kiện mình làm dở: sự kiện từ GitHub, Jira và hệ thống deploy, được gom theo window để tính các chỉ số như thời gian review PR hay tần suất deploy. Phần về việc dữ liệu có thể mất hoặc trùng ở đâu trong cùng pipeline nằm ở bài exactly-once không tự có.

Event time và watermark

Một PR được mở lúc 10:00 nhưng webhook có thể đến lúc 10:03. Muốn tính "bao nhiêu PR mở trong giờ 10" cho đúng, window phải dựa trên event time (thời điểm sự việc xảy ra), không phải thời điểm Flink nhận được.

Vấn đề là Flink không biết khi nào đã nhận đủ sự kiện của giờ 10. Watermark là cách nói "từ giờ trở đi, mình không chờ sự kiện nào cũ hơn thời điểm T nữa". Cách phổ biến:

WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofMinutes(5))
    .withTimestampAssigner((e, ts) -> e.eventTime().toEpochMilli())
    .withIdleness(Duration.ofSeconds(30));

Watermark luôn đi sau sự kiện mới nhất 5 phút. Window giờ 10 chỉ đóng khi watermark vượt 11:00, tức khi đã thấy sự kiện từ 11:05 trở đi. Đánh đổi rất rõ: chờ lâu hơn thì ít sự kiện muộn bị bỏ sót hơn, nhưng kết quả ra chậm hơn.

Sự kiện quá muộn bị bỏ, và không ai báo

Có thể cho window thêm thời gian sau khi đóng bằng allowedLateness: sự kiện đến trong khoảng đó sẽ cập nhật lại kết quả của window. Nhưng sự kiện đến sau cả khoảng đó thì sao?

Mặc định, Flink bỏ đi mà không báo gì. Tài liệu thiết kế của dự án ghi "sự kiện quá muộn đi vào side output rồi vào DLQ", nhưng code chưa hề có dòng đó. Cách làm đúng chỉ tốn vài dòng:

OutputTag<Event> lateTag = new OutputTag<>("late-events") {};

var result = stream
    .keyBy(Event::teamId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .allowedLateness(Time.minutes(5))
    .sideOutputLateData(lateTag)
    .aggregate(new CountAgg());

result.getSideOutput(lateTag).sinkTo(lateEventsSink);   // và đếm bằng một metric

Không cần xử lý sự kiện muộn ngay, nhưng phải đếm được chúng. Con số đó cho biết 5 phút có đủ hay không, thay vì đoán.

Số phận của một sự kiện do watermark lúc nó đến quyết định, không phải timestamp của chính nó.

Một phân vùng im lặng có thể làm cả job đứng yên

Với nguồn Kafka nhiều phân vùng, watermark của nguồn là watermark nhỏ nhất trong các phân vùng. Nếu một phân vùng không có dữ liệu, watermark của nó không tiến, và kéo cả job đứng lại: không window nào đóng.

withIdleness(30s) đánh dấu một phân vùng là "rảnh" khi nó im lặng 30 giây, và Flink bỏ qua nó khi tính watermark chung.

Chỉ một phân vùng trống là đủ giữ giá trị nhỏ nhất đứng yên, kéo theo mọi window trong job.

Tài liệu thiết kế ban đầu chỉ ghi withIdleness cho chế độ chạy thật, còn chế độ replay (đọc lại dữ liệu lịch sử) chỉ cần timestamp tăng dần, vì dữ liệu lịch sử đã có thứ tự. Trong code có một comment sửa lại điều đó: khi replay, chỉ vài trong số nhiều phân vùng có dữ liệu, các phân vùng trống giữ watermark ở Long.MIN_VALUE, và không window nào đóng. Job chạy, đọc hết dữ liệu, mà không ra kết quả nào. Đây là kiểu lỗi chỉ thấy khi chạy thử, không thấy khi đọc thiết kế.

Cần biết thêm mặt trái của withIdleness: khi một phân vùng rảnh có dữ liệu trở lại, các sự kiện đầu tiên của nó có thể đã cũ hơn watermark chung và bị coi là muộn. Đặt thời gian rảnh quá ngắn sẽ biến sự chậm trễ bình thường thành mất dữ liệu.

State TTL: ba chi tiết dễ hiểu sai

State như tập khoá đã khử trùng cần hết hạn để không phình mãi. Cấu hình trong dự án:

StateTtlConfig.newBuilder(Time.hours(24))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter(1000)
    .build();

1. OnCreateAndWrite chứ không phải OnReadAndWrite. Với state khử trùng, mỗi sự kiện đều đọc state để kiểm tra. Nếu đọc cũng gia hạn TTL, một khoá bị gửi trùng liên tục sẽ không bao giờ hết hạn. Gia hạn chỉ khi ghi là đúng ý: khoá sống 24 giờ kể từ lần đầu thấy.

2. Tham số của cleanupInRocksdbCompactFilter không phải tỉ lệ lấy mẫu. Comment trong code ghi "1000 nghĩa là kiểm tra 1 trong mỗi 1000 khoá khi compaction". Thực ra mọi khoá đều được kiểm tra; con số này là số entry được xử lý trước khi bộ lọc hỏi lại thời gian hiện tại từ Flink. Số nhỏ thì dọn chính xác hơn nhưng compaction chậm hơn, vì mỗi lần hỏi là một lời gọi JNI. Nên đọc tài liệu API thay vì tin comment, kể cả comment do chính mình viết.

3. TTL dựa trên đồng hồ thật, không phải event time. Theo tài liệu của Flink, state TTL hiện chỉ hỗ trợ processing time. Hệ quả: khi replay dữ liệu 30 ngày trong một giờ, state 24 giờ sẽ không hết hạn như lúc chạy thật. Điều này dẫn tới phần tiếp theo.

Replay "tất định" và những thứ ngầm phá vỡ nó

Một lời hứa hấp dẫn của event time: cùng dữ liệu vào, cùng kết quả ra, bất kể chạy lúc nào, nhanh hay chậm. Nhờ đó có thể sửa logic rồi chạy lại lịch sử để tính lại chỉ số. Tài liệu thiết kế của dự án cũng hứa như vậy, với điều kiện "không có logic phụ thuộc processing time".

Đọc lại code thì thấy ba chỗ cần canh, hai trong số đó ngầm vi phạm lời hứa này:

Timer theo processing time. Pipeline tính chỉ số theo cửa sổ 30 ngày đặt một timer cho "đầu ngày hôm sau" của sự kiện bằng registerProcessingTimeTimer. Khi chạy thật, timer kêu lúc ngày đã qua, sau khi mọi sự kiện của ngày đó đã vào. Khi replay, mốc đó đã nằm trong quá khứ so với đồng hồ thật, nên timer kêu gần như ngay lập tức, lúc mới có vài sự kiện đầu của ngày. Kết quả phụ thuộc vào tốc độ đọc. Cách đúng là registerEventTimeTimer: timer kêu khi watermark qua mốc, nên chạy nhanh hay chậm đều ra cùng chuỗi kết quả.
Instant.now() trong kết quả. Bản ghi kết quả có một trường lấy từ đồng hồ thật. Ở đây nó chỉ là siêu dữ liệu "tính lúc nào" nên vô hại, nhưng đây là chỗ cần canh: chỉ cần giá trị đó tham gia khoá hay version là mỗi lần chạy lại sẽ ra một dòng khác.
Cờ replay không nằm trong khoá. Bảng ClickHouse dùng ReplacingMergeTree với cột is_replay để tách dữ liệu chạy lại khỏi dữ liệu thật. Nhưng is_replay không nằm trong ORDER BY. Với engine này, các dòng cùng khoá sắp xếp được coi là một, nên khi merge, dòng replay có thể thay thế dòng thật. Muốn tách hai loại dữ liệu trong cùng bảng, cột phân biệt phải là một phần của khoá, hoặc tách hẳn ra bảng khác.

Quy tắc mình rút ra: trong mọi operator, chỉ được dùng thời gian lấy từ sự kiện hoặc từ watermark. Bất cứ chỗ nào gọi đồng hồ hệ thống đều cần một lý do rõ ràng.

Bonus: "ngoài giờ" là ngoài giờ của ai?

Cùng pipeline đó tính tín hiệu "làm việc ngoài giờ" bằng giờ UTC:

int hour = e.eventTime().atZone(ZoneOffset.UTC).getHour();
return hour < 9 || hour >= 18;

Với một người ở Hà Nội (UTC+7), khung 9h–18h UTC là 16h tới 1h sáng. Cả buổi sáng làm việc bình thường bị tính là "ngoài giờ". Event time cho biết sự việc xảy ra khi nào; còn "giờ hành chính" là khái niệm của từng người, phải tính theo múi giờ của người đó. Đây là cùng một loại bẫy với việc âm lịch Việt Nam tính theo UTC+7: thời điểm là tuyệt đối, nhưng "ngày" và "giờ" thì không.

Tóm tắt

Window theo event time, watermark chậm hơn sự kiện mới nhất một khoảng vừa đủ.
Luôn sideOutputLateData và đếm sự kiện muộn; mặc định Flink bỏ chúng mà không báo.
withIdleness cần cho cả chạy thật lẫn replay, và đừng đặt quá ngắn.
State TTL: gia hạn khi ghi cho state khử trùng, đọc kỹ ý nghĩa tham số dọn dẹp, và nhớ TTL chạy theo đồng hồ thật.
Muốn replay tất định: timer theo event time, không dùng đồng hồ hệ thống trong logic, và cờ phân biệt phải nằm trong khoá.

Thời gian trong stream processing còn nhiều ngóc ngách mình chưa gặp. Nếu bạn có kinh nghiệm hay, mình rất muốn được trao đổi. Email của mình ở trang chủ.