Flink 4 — Quản lý trạng thái & Checkpoint

Vì sao stateful processing là siêu năng lực của Flink

Nếu chỉ xử lý từng sự kiện độc lập — lọc, biến đổi, định tuyến — thì bất kỳ công cụ nào cũng làm được, và bạn không thực sự cần Flink. Sức mạnh thật của Flink nằm ở stateful processing: khả năng nhớ thông tin giữa các sự kiện. Một toán tử có thể giữ trong bộ nhớ của nó một "sổ ghi nhớ" và cập nhật liên tục khi từng bản ghi chảy qua.

Nhớ được giữa các sự kiện mở ra gần như toàn bộ các bài toán stream có giá trị:

• Đếm luỹ kế / tổng hợp liên tục: "tài khoản này đã có bao nhiêu giao dịch trong 5 phút qua", "tổng số tiền chuyển hôm nay". Không nhớ số đếm trước đó thì không cộng dồn được.
• Deduplication (dedup): nhớ tập transactionId đã thấy để bỏ qua bản ghi trùng do Kafka gửi lại hoặc CDC replay.
• Join hai luồng: nhớ các bản ghi bên A chưa gặp cặp bên B (ví dụ ghép sự kiện "khởi tạo giao dịch" với "xác thực OTP" đến sau vài giây).
• Phát hiện mẫu (pattern / CEP): nhớ chuỗi trạng thái đang diễn ra để nhận ra "3 lần nhập sai PIN rồi rút tiền" — một chuỗi sự kiện, không phải một sự kiện đơn lẻ.

Bài này nối tiếp mô hình dataflow và event time & windows — bản thân window cũng là một dạng state được Flink quản lý sẵn. Ở đây ta đi xuống lớp bên dưới: state được lưu ở đâu, được chụp lại (checkpoint) thế nào để job không mất trí nhớ sau sự cố, và làm sao đạt exactly-once.

Điều làm Flink khác biệt: state là công dân hạng nhất, được framework quản lý — phân vùng theo key, sao lưu định kỳ, khôi phục tự động khi lỗi. Bạn viết logic nghiệp vụ, Flink lo phần "trí nhớ bền bỉ có khả năng chịu lỗi".

Hai họ state: keyed state vs operator state

Keyed state — trạng thái theo khoá

Đây là loại được dùng nhiều nhất. Sau khi gọi keyBy(accountId), dòng dữ liệu được phân vùng theo key, và mỗi key có một bản state riêng, tách biệt hoàn toàn với key khác. Toán tử xử lý tài khoản A không bao giờ thấy state của tài khoản B. Điều này cho phép Flink phân tán state qua nhiều máy theo key và song song hoá vô tư.

Flink cung cấp sẵn các kiểu keyed state, chọn theo hình dạng dữ liệu cần nhớ:

Điểm mấu chốt về hiệu năng: ListState, MapState, ReducingState, AggregatingState cho phép cập nhật tăng dần (incremental) — thêm một phần tử, cập nhật một entry — mà không phải đọc/ghi lại toàn bộ. Với RocksDB backend, dùng MapState thay vì nhét một HashMap khổng lồ vào ValueState là khác biệt sống còn về hiệu năng, vì ValueState phải serialize/deserialize cả object mỗi lần chạm tới.

// Minh hoạ (Flink DataStream API, Java — KHÔNG phải SQL chạy được trong sandbox):
// đếm luỹ kế giao dịch theo tài khoản, phát cảnh báo khi vượt ngưỡng
public class CountAlert extends KeyedProcessFunction<String, Transaction, Alert> {
    private transient ValueState<Long> countState;

    @Override
    public void open(Configuration cfg) {
        ValueStateDescriptor<Long> d =
            new ValueStateDescriptor<>("txCount", Long.class);
        countState = getRuntimeContext().getState(d);   // state gắn theo key
    }

    @Override
    public void processElement(Transaction tx, Context ctx, Collector<Alert> out) throws Exception {
        Long cur = countState.value();
        long next = (cur == null ? 0L : cur) + 1;
        countState.update(next);                          // cập nhật state của key này
        if (next > 20) out.collect(new Alert(tx.getAccountId(), next));
    }
}

Operator state — trạng thái theo instance toán tử

Không phân theo key mà gắn với mỗi instance song song của toán tử. Ít gặp trong code nghiệp vụ, chủ yếu dùng khi viết source/sink. Ví dụ kinh điển: Kafka source nhớ offset đang đọc của từng partition — đó là operator state. Khi thay đổi độ song song (rescale), Flink phải phân phối lại operator state; có hai kiểu phân phối: even-split redistribution (chia đều danh sách phần tử, như offset partition) và union redistribution (mỗi instance nhận toàn bộ, dùng cho state broadcast).

Ngoài ra còn broadcast state: phát cùng một tập dữ liệu (ví dụ bảng luật gian lận cập nhật động) tới mọi instance để mỗi bản ghi được đối chiếu với luật mới nhất.

State backend: nơi state thực sự nằm

State được lưu ở đâu trong lúc job chạy? Đó là vai trò của state backend. Flink hiện có hai lựa chọn chính, và chọn sai là một trong những nguyên nhân phổ biến khiến job hết RAM hoặc chậm.

HashMapStateBackend giữ mọi state dưới dạng object Java trên heap — cực nhanh vì không có serialize, nhưng toàn bộ state phải vừa trong RAM và làm tăng áp lực garbage collection. Phù hợp khi state gọn: vài triệu key nhỏ, window ngắn.

EmbeddedRocksDBStateBackend lưu state trong RocksDB — một key-value store nhúng ghi xuống đĩa cục bộ. State có thể lớn hơn RAM nhiều lần vì đĩa lớn hơn RAM. Đổi lại mỗi lần đọc/ghi state phải serialize/deserialize và có thể chạm đĩa, nên chậm hơn. Ưu điểm lớn nhất về vận hành: incremental checkpoint — chỉ chụp các file SST mới thay đổi kể từ checkpoint trước, thay vì chụp lại toàn bộ state. Với state hàng trăm GB, đây là khác biệt giữa checkpoint vài giây và checkpoint vài phút.

Nguyên tắc chọn cho ngân hàng: nếu bạn giữ state theo từng tài khoản (hàng chục triệu tài khoản, mỗi tài khoản có bộ đếm + lịch sử ngắn) → RocksDB gần như bắt buộc, vì tổng state vượt xa RAM một TaskManager. Nếu chỉ dedup theo cửa sổ vài phút với vài trăm nghìn key → HashMap nhanh hơn.

Checkpoint — chụp state nhất quán để chịu lỗi

State trong RAM/đĩa cục bộ sẽ mất sạch nếu TaskManager chết. Nếu không có cơ chế sao lưu, một sự cố phần cứng sẽ xoá toàn bộ số đếm, buffer join, tiến độ dedup — job phải bắt đầu lại từ số 0. Không chấp nhận được với nghiệp vụ ngân hàng.

Checkpoint là ảnh chụp (snapshot) nhất quán toàn cục của toàn bộ state trong job, được Flink chụp tự động, định kỳ, và lưu ra kho bền (HDFS/S3/GCS). Khi có sự cố, Flink khởi động lại toàn bộ job từ checkpoint gần nhất và tiếp tục như chưa hề có chuyện gì.

Vì sao "nhất quán" là bài toán khó — thuật toán barrier

Điểm khó: job có nhiều toán tử chạy song song trên nhiều máy, dữ liệu đang chảy giữa chúng. Nếu ta chụp mỗi toán tử ở một thời điểm khác nhau, các ảnh chụp không nhất quán — có thể một sự kiện đã được toán tử A tính nhưng chưa tới B, khi khôi phục sẽ bị đếm hai lần hoặc mất.

Flink giải bằng biến thể của thuật toán Chandy–Lamport (chụp trạng thái phân tán) dùng checkpoint barrier. Cơ chế:

1. JobManager (checkpoint coordinator) định kỳ tiêm một barrier mang số hiệu n vào tất cả các source.
2. Barrier chảy theo dòng dữ liệu cùng các bản ghi. Khi một toán tử nhận barrier n từ tất cả các đầu vào, nó chụp state của mình ra kho bền rồi chuyển tiếp barrier n xuống downstream.
3. Khi mọi toán tử (kể cả sink) đã ghi xong ảnh chụp cho barrier n, checkpoint n được coi là hoàn tất (completed) và bền vững.

Cái đẹp là barrier tách dòng thành "trước barrier" (đã nằm trong ảnh chụp) và "sau barrier" (chưa) — đảm bảo mọi toán tử chụp cùng một ranh giới logic của dòng, cho ảnh nhất quán mà không cần dừng xử lý.

Barrier alignment — cái giá của nhất quán

Với toán tử có nhiều đầu vào (sau keyBy, union, join), barrier từ các luồng đến không đồng thời. Toán tử phải chờ nhận đủ barrier n từ mọi đầu vào rồi mới chụp — gọi là barrier alignment. Trong lúc chờ, luồng nào đã có barrier n sẽ bị chặn tạm (buffer lại), gây thêm độ trễ. Khi backpressure nặng, alignment có thể kéo dài và làm checkpoint chậm/timeout.

Flink có unaligned checkpoint: thay vì chờ, toán tử chụp luôn cả các bản ghi đang nằm trong buffer (in-flight) vào ảnh chụp. Đổi lại checkpoint nhanh hơn dưới backpressure nhưng kích thước ảnh chụp lớn hơn. Đây là công tắc vận hành quan trọng khi job hay bị nghẽn.

Các tham số checkpoint cần chỉnh

• Checkpoint interval: khoảng cách giữa hai checkpoint (ví dụ mỗi 60 giây). Ngắn → mất ít dữ liệu phải replay khi lỗi (RPO thấp) nhưng tốn tài nguyên I/O; dài → nhẹ hơn nhưng khôi phục phải xử lý lại nhiều dữ liệu hơn.
• Checkpoint timeout: nếu một checkpoint không hoàn tất trong khoảng này (thường vì backpressure), nó bị huỷ.
• Min pause between checkpoints: khoảng nghỉ tối thiểu giữa các checkpoint, tránh checkpoint chồng lên nhau bào mòn hiệu năng.
• Mode: EXACTLY_ONCE (mặc định, dùng alignment) hoặc AT_LEAST_ONCE (không align, nhanh hơn nhưng có thể xử lý lại).

// Minh hoạ cấu hình checkpoint (không phải SQL chạy được)
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new EmbeddedRocksDBStateBackend(true)); // true = incremental
env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE); // mỗi 60s
env.getCheckpointConfig().setCheckpointTimeout(120_000);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // giữ checkpoint khi huỷ job

Savepoint — ảnh chụp thủ công để nâng cấp job

Checkpoint phục vụ máy (tự động khôi phục khi lỗi, do Flink quản lý và dọn dẹp). Savepoint phục vụ con người: một ảnh chụp state được kích hoạt thủ công, ở định dạng bền vững và có kiểm soát phiên bản, để bạn chủ động dừng job rồi khởi động lại từ đúng điểm đó.

Savepoint là công cụ vận hành hằng ngày: nâng cấp phiên bản Flink, sửa/deploy logic mới cho job, rescale (đổi độ song song), di chuyển job sang cluster khác — tất cả đều theo quy trình stop-with-savepoint rồi start-from-savepoint, không mất state. Không có savepoint, mỗi lần sửa code là mất sạch trí nhớ của job.

# Minh hoạ CLI: chụp savepoint rồi dừng job (để deploy phiên bản mới)
flink stop --savepointPath s3://ncb-flink/savepoints <jobId>
# Khởi động lại job (đã sửa code) từ savepoint
flink run -s s3://ncb-flink/savepoints/savepoint-abc <jar>

State TTL — tự dọn state cũ

State keyed tăng theo số key. Nếu key là accountId cố định thì ổn, nhưng nếu key là transactionId (dedup) hay sessionId thì state phình vô hạn — cuối cùng làm chậm checkpoint và hết đĩa. State TTL (time-to-live) cho phép Flink tự động xoá state không được chạm tới sau một khoảng thời gian.

// Minh hoạ: dedup transactionId trong 24h, sau đó tự xoá
StateTtlConfig ttl = StateTtlConfig.newBuilder(Time.hours(24))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) // reset TTL khi ghi
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter(1000) // dọn khi RocksDB compaction
    .build();
descriptor.enableTimeToLive(ttl);

Lưu ý: TTL dựa trên processing time, không phải event time. Chọn UpdateType (chỉ reset khi tạo, hay reset cả khi đọc/ghi) và chiến lược dọn (nền, khi truy cập, hay khi RocksDB compaction) ảnh hưởng tới độ chính xác kích thước state. Với dedup, TTL 24–48h thường đủ vì bản ghi trùng hiếm khi cách nhau lâu hơn thế.

Exactly-once dựa trên checkpoint + sink giao dịch

Checkpoint đảm bảo state nội bộ của Flink là exactly-once: khi khôi phục, mọi bộ đếm/buffer trở về đúng ranh giới barrier, không xử lý lặp về mặt state. Nhưng exactly-once đầu-cuối (end-to-end) còn cần sink phối hợp: sau khi khôi phục, Flink sẽ replay các bản ghi từ sau checkpoint gần nhất — nếu sink đã ghi những bản ghi đó ra ngoài rồi thì bị trùng.

Lời giải là two-phase commit (2PC): sink ghi kết quả ở trạng thái "chưa commit", và chỉ commit khi checkpoint tương ứng hoàn tất. Nếu lỗi trước khi checkpoint xong, phần ghi dở bị huỷ (abort) khi khôi phục. Kafka sink dùng transaction của Kafka; JDBC/file sink dùng cơ chế transaction/staged file tương tự. Đầu vào cũng phải "tua lại được" (replayable) — Kafka nhớ offset trong operator state đáp ứng điều này.

Đây là lý do một pipeline exactly-once thực thụ đòi hỏi cả ba: nguồn replayable + checkpoint (barrier) + sink giao dịch. Chủ đề này được đào sâu ở bài chịu lỗi & exactly-once.

Schema evolution — state sống lâu hơn một phiên bản code

State của một job có thể tồn tại hàng tháng qua nhiều lần deploy. Khi bạn sửa class của state (thêm trường vào object đếm, đổi kiểu), Flink phải đọc được state cũ đã lưu bằng schema mới — gọi là state schema evolution. Flink hỗ trợ tiến hoá schema cho các kiểu như POJO và Avro: thêm/bớt trường thường an toàn (trường mới nhận giá trị mặc định), nhưng đổi kiểu trường hoặc đổi key thì không tương thích và có thể làm khôi phục từ savepoint thất bại. Nguyên tắc: dùng serializer ổn định (Avro/POJO), tránh Kryo cho state lâu dài, và thử khôi phục từ savepoint ở môi trường staging trước khi deploy thật.

Use case thực tế: phát hiện gian lận theo ngưỡng, khôi phục không mất đếm

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 vượt 20 giao dịch trong 5 phút — dấu hiệu bot rút tiền hoặc dò thẻ. Hệ thống chạy 24/7, xử lý ~8.000 giao dịch/giây giờ cao điểm, giữ state cho ~15 triệu tài khoản đang hoạt động. Yêu cầu vận hành cứng: khi TaskManager chết hoặc khi deploy luật mới, không được reset số đếm về 0 — nếu không, một tài khoản đang tấn công sẽ "được xoá lịch sử" và lọt lưới đúng lúc nguy hiểm.

Thiết kế state:

1. Keyed state theo accountId: mỗi tài khoản giữ một AggregatingState (số đếm giao dịch trong cửa sổ trượt) + MapState<phút, count> để cộng/bớt theo phút. Dùng MapState để cập nhật tăng dần, tránh serialize lại cả object mỗi bản ghi.
2. State backend = RocksDB (incremental): 15 triệu tài khoản × state nhỏ vẫn vượt RAM một TaskManager → RocksDB. Incremental checkpoint giữ thời gian chụp ở mức vài giây dù tổng state ~hàng chục GB.
3. Checkpoint interval 60s, EXACTLY_ONCE: nếu TaskManager chết, khôi phục từ checkpoint gần nhất → mất tối đa ~60s dữ liệu phải replay từ Kafka (nguồn replayable qua offset), số đếm mỗi tài khoản trở về đúng giá trị tại ranh giới checkpoint, không về 0.
4. State TTL 6 giờ trên state phụ trợ (buffer dedup theo transactionId) để không phình vô hạn; state đếm chính gắn accountId cố định nên không cần TTL.
5. Sink cảnh báo giao dịch (Kafka 2PC): cảnh báo chỉ commit khi checkpoint hoàn tất → sau khôi phục không phát trùng một cảnh báo gian lận (tránh làm phiền đội điều tra bằng alert lặp).
6. Nâng cấp luật bằng savepoint: khi đổi ngưỡng 20 → 15, chạy stop-with-savepoint rồi start-from-savepoint với jar mới. Số đếm luỹ kế của mọi tài khoản được giữ nguyên qua lần deploy.

Kết quả (minh hoạ tình huống). Trong một đợt diễn tập sự cố: kill một TaskManager giữa cao điểm. Flink phát hiện, khôi phục từ checkpoint 47 giây trước đó, replay ~370.000 giao dịch từ Kafka trong ~12 giây, và mọi bộ đếm tài khoản khớp chính xác với trạng thái trước sự cố — 3 tài khoản đang bị tấn công vẫn giữ số đếm 14/16/18 (không về 0), và cảnh báo bắn đúng khi chúng vượt 20 vài giây sau đó. Con số minh hoạ để làm rõ giá trị của checkpoint, không phải benchmark chính thức.

Logic "đếm giao dịch theo ngưỡng để phát hiện hành vi đáng ngờ" này gắn trực tiếp với các luật trong giám sát giao dịch AML — nơi việc không mất trí nhớ khi hệ thống sự cố quyết định trực tiếp tới việc có bỏ lọt hành vi rửa tiền hay không.

Ghi nhớ

• Stateful processing là lý do tồn tại của Flink: nhớ giữa các sự kiện để đếm luỹ kế, dedup, join luồng, phát hiện mẫu. Không nhớ thì không cần Flink.
• Keyed state (theo key sau keyBy, mỗi key tách biệt): ValueState (một giá trị), ListState, MapState, ReducingState, AggregatingState. Ưu tiên MapState thay vì nhét HashMap vào ValueState để cập nhật tăng dần. Operator state gắn theo instance toán tử, chủ yếu cho source/sink (ví dụ offset Kafka).
• State backend: HashMapStateBackend (heap, nhanh, giới hạn RAM, GC nặng) vs EmbeddedRocksDBStateBackend (đĩa, state lớn hơn RAM, có incremental checkpoint). State theo hàng triệu tài khoản → RocksDB.
• Checkpoint = ảnh chụp nhất quán toàn cục, tự động, định kỳ, để khôi phục sau lỗi. Dùng barrier (Chandy–Lamport) để chụp cùng ranh giới logic mà không dừng job; toán tử nhiều đầu vào cần barrier alignment (đánh đổi độ trễ). Chỉnh interval/timeout/min-pause; unaligned checkpoint khi backpressure nặng.
• Savepoint = ảnh chụp thủ công để nâng cấp/di chuyển/rescale/sửa job (stop-with-savepoint → start-from-savepoint). Khác checkpoint ở mục đích và vòng đời (người dùng tự quản).
• State TTL tự xoá state cũ (theo processing time) — bắt buộc khi key vô hạn như transactionId/sessionId để tránh phình state.
• Exactly-once đầu-cuối = nguồn replayable + checkpoint + sink giao dịch (two-phase commit) commit theo checkpoint. Chi tiết ở bài chịu lỗi.
• Schema evolution: state sống lâu hơn code; thêm/bớt trường (POJO/Avro) thường an toàn, đổi kiểu thì không. Thử khôi phục từ savepoint ở staging trước khi deploy.