[DE Blog #11] Thiết Kế Hệ Thống Phát Hiện Gian Lận Thời Gian Thực (Real-Time Fraud Detection) Với Apache Flink, Stateful Streaming & Feature Store

1. Bối cảnh thực tế (Context & Problem Statement)

Trong các hệ thống thanh toán ngân hàng (Banking), ví điện tử hoặc sàn thương mại điện tử, việc phát hiện gian lận sau khi giao dịch đã hoàn tất (Post-transaction) sẽ gây thất thoát tài chính nặng nề. Do đó, yêu cầu đặt ra là: Phải đánh giá mức độ rủi ro và quyết định Chấp thuận (Approve) hay Chặn (Block) giao dịch trong vòng dưới $100\text{ms}$ ngay trong luồng thanh toán.

Thách thức kỹ thuật khổng lồ cho Data Engineer:

Làm thế nào để tính toán các đặc trưng phức tạp theo thời gian thực (ví dụ: “Số lần quẹt thẻ trong 5 phút qua”, “Tổng số tiền chi tiêu trong 1 giờ qua tại 2 thành phố cách nhau 1000km”) với hàng chục nghìn giao dịch mỗi giây?
Làm thế nào để duy trì trạng thái lịch sử (State) khổng lồ hàng Terabyte trên bộ nhớ phân tán mà không bị sập và vẫn đảm bảo Exactly-Once State Consistency khi có máy chủ bị chết đột ngột?

2. Các Khái Niệm & Cơ Chế Cốt Lõi

2.1. Stateful Stream Processing Với Apache Flink

Stateless Streaming: Mỗi bản ghi được xử lý độc lập hoàn toàn (ví dụ: parse JSON, filter IP).
Stateful Streaming: Quá trình xử lý một bản ghi phụ thuộc vào thông tin lịch sử của các bản ghi trước đó (ví dụ: tính tổng tiền luân phiên, đếm số lần đăng nhập sai liên tiếp).
Flink State Backends (Nơi lưu trữ State):
- HashMapStateBackend (In-Memory): Lưu state trực tiếp trên Java Heap. Tốc độ truy xuất cực nhanh ( $< 1\mu\text{s}$ ) nhưng bị giới hạn bởi dung lượng RAM của TaskManager và chịu chi phí Garbage Collection (GC) lớn.
- EmbeddedRocksDBStateBackend: Lưu state trên ổ đĩa cục bộ (Local NVMe SSD) thông qua cơ sở dữ liệu nhúng RocksDB. Cho phép lưu trữ State khổng lồ vượt quá dung lượng RAM (hàng Terabyte), chỉ tải các block dữ liệu thường dùng vào bộ nhớ đệm (Off-Heap).

2.2. Cơ Chế Checkpointing & Thuật Toán Chandy-Lamport (Asynchronous Barrier Snapshotting)

Để đảm bảo tính nhất quán trạng thái Exactly-Once, Flink không dừng luồng xử lý mà chèn các điểm mốc (Checkpoint Barriers) xen kẽ vào dòng dữ liệu:

JobManager định kỳ gửi một Barrier $N$ vào các nguồn dữ liệu (Kafka Sources).
Barrier di chuyển theo dòng chảy dữ liệu qua từng Operator. Khi một Operator nhận đủ Barrier từ tất cả các kênh đầu vào, nó sẽ chụp lại ảnh trạng thái (State Snapshot) và ghi bất đồng bộ ra Cloud Storage (S3/GCS).
Luồng xử lý dữ liệu chính vẫn tiếp tục chạy bình thường mà không bị block. Nếu có node sập, toàn bộ pipeline chỉ cần khôi phục lại trạng thái từ Checkpoint thành công gần nhất.

Source -> [Data 3] [Data 2] [BARRIER 1] [Data 1] -> Operator (Compute & Save State)

2.3. Khái Niệm Feature Store Trong Data & Machine Learning

Feature Store là lớp quản lý và phục vụ các biến số đặc trưng (Features) tập trung:

Dual-Storage Architecture:
- Online Feature Store (Redis, DynamoDB, Aerospike): Lưu trữ giá trị feature mới nhất, tối ưu cho việc đọc với độ trễ siêu thấp ( $< 5\text{ms}$ ) phục vụ trực tiếp cho mô hình AI / Rule Engine chấm điểm giao dịch.
- Offline Feature Store (Delta Lake, Apache Iceberg, Snowflake): Lưu trữ toàn bộ lịch sử biến động của feature theo thời gian, phục vụ việc huấn luyện mô hình (Model Training).
Point-in-Time Correctness (Time-Travel Join): Đảm bảo khi huấn luyện mô hình trên dữ liệu quá khứ, mô hình chỉ được nhìn thấy các giá trị feature tại đúng thời điểm giao dịch xảy ra, loại bỏ hoàn toàn hiện tượng Rò rỉ dữ liệu tương lai (Data Leakage / Train-Serve Skew).

3. Các Bảng Markdown So Sánh Chi Tiết

📊 Bảng 1: So Sánh Apache Spark Structured Streaming vs. Apache Flink

Tiêu chí	Apache Spark Structured Streaming	Apache Flink
Mô hình xử lý	Micro-batching (gom dữ liệu thành các batch nhỏ vài trăm ms)	Native Streaming (Event-driven) (xử lý từng sự kiện đơn lẻ ngay khi đến)
Độ trễ xử lý (Latency)	Thấp ( $100\text{ms} - 500\text{ms}$ )	Cực thấp (Sub-millisecond: $< 10\text{ms}$ )
Quản lý trạng thái (State)	Tương đối nặng khi state lớn (State Store trên HDFS/RocksDB)	Xuất sắc nhất thế giới (Tối ưu hóa sâu với RocksDB State Backend)
Xử lý Event Time & Watermark	Tốt	Hoàn hảo (Hỗ trợ cực mạnh cho Out-of-order data và Session Windows)
Hệ sinh thái đi kèm	Rất mạnh về Batch, SQL, MLlib trên cùng 1 nền tảng	Tập trung chuyên sâu vào Real-time Streaming, CEP và Stateful Processing
Use-case tối ưu	ETL gần thời gian thực, tổng hợp số liệu nạp Data Lakehouse	Fraud Detection, Real-Time Alerting, Dynamic Pricing, IoT

📊 Bảng 2: So Sánh HashMapStateBackend vs. EmbeddedRocksDBStateBackend Trong Flink

Tiêu chí	HashMapStateBackend	EmbeddedRocksDBStateBackend
Vị trí lưu trữ	On-Heap JVM Memory (RAM)	Out-of-Core (Local SSD/NVMe + Off-Heap Cache)
Dung lượng State tối đa	Giới hạn theo dung lượng RAM khả dụng	Không giới hạn (vượt dung lượng RAM, phụ thuộc vào dung lượng ổ đĩa)
Tốc độ đọc/ghi State	Nhanh nhất (truy cập con trỏ Java thuần túy)	Chậm hơn một chút (phải serialize/deserialize dữ liệu qua JNI C++)
Tác động Garbage Collection	Có thể gây lag do GC Pause khi state lớn	Gần như không bị ảnh hưởng bởi JVM GC
Incremental Checkpointing	Không hỗ trợ (phải lưu toàn bộ state mỗi lần)	Có hỗ trợ (chỉ lưu phần SST files thay đổi $\rightarrow$ checkpoint cực nhanh)
Khuyến nghị sử dụng	State nhỏ ( $< 10\text{GB}$ ), cần độ trễ sub-millisecond	Production quy mô lớn, State từ hàng trăm GB đến nhiều TB

📊 Bảng 3: So Sánh Online Feature Store vs. Offline Feature Store

Tiêu chí	Online Feature Store	Offline Feature Store
Mục đích chính	Phục vụ suy luận mô hình thời gian thực (Real-time Inference)	Huấn luyện mô hình (Model Training), Phân tích Batch
Công nghệ tiêu biểu	Redis, AWS DynamoDB, Aerospike, Cassandra	Apache Iceberg, Delta Lake, Snowflake, BigQuery
Dữ liệu lưu trữ	Chỉ lưu giá trị mới nhất (Latest State) của từng Entity	Lưu trữ toàn bộ lịch sử (Historical Snapshots)
Yêu cầu độ trễ	$< 5\text{ms}$ (P99)	Vài giây đến vài phút (tối ưu hóa quét hàng triệu dòng)
Phương thức truy cập	Key-Value Lookup (`get_features(user_id='123')`)	SQL Queries, Spark Batch DataFrames

4. Kiến Trúc Thiết Kế Toàn Diện (Full System Design Walkthrough)

[1. User Transaction] 
         | (HTTP POST /pay)
         v
+-----------------------------------------------------------------------------------+
| SYNCHRONOUS EVALUATION SERVICE (Latency SLA < 50ms)                               |
|                                                                                   |
|  [Payment Gateway] ---> [Fraud Scoring Engine] ---> [Decision: APPROVE / BLOCK]   |
|                               ^                                                   |
|                               | (1) Read features (< 5ms)                         |
|                               v                                                   |
|                    [Online Feature Store (Redis)]                                 |
+-----------------------------------------------------------------------------------+
         |
         | (2) Async Event Log
         v
+-----------------------------------------------------------------------------------+
| ASYNCHRONOUS REAL-TIME FEATURE PIPELINE (Apache Flink)                           |
|                                                                                   |
|  [Kafka: "payment-events"] ---> [Apache Flink Job (Stateful Engine)]              |
|                                    - RocksDB State (Rolling 5m, 1h aggregates)    |
|                                    - Sliding Windows & Pattern Detection          |
|                                           |                                       |
|                                           +---> (3) Write features -> [Redis]     |
|                                           |                                       |
|                                           +---> (4) Raw Storage -> [Iceberg/S3]   |
+-----------------------------------------------------------------------------------+
                                                                   |
                                                                   v
                                                  [Offline Store (Model Retraining)]

Quy trình hoạt động 4 bước:

Giao dịch đồng bộ (Sync Evaluation): Người dùng thực hiện thanh toán $\rightarrow$ Payment API gọi sang Fraud Scoring Service. Service này truy vấn nhanh các đặc trưng của user (ví dụ: user_tx_count_5min, avg_amount_1h) từ Redis (Online Feature Store) trong $< 5\text{ms}$ và nạp vào Model ML để ra quyết định Chấp thuận/Từ chối.
Phát sự kiện bất đồng bộ: Giao dịch được đẩy vào Apache Kafka Topic payment-events.
Tính toán đặc trưng động (Async Computation): Apache Flink tiêu thụ dòng sự kiện từ Kafka, cập nhật các cửa sổ thời gian trượt (Sliding Windows) được lưu trong RocksDB, và ghi đè các chỉ số mới nhất ngược lại vào Redis cho các giao dịch kế tiếp sử dụng.
Lưu trữ & Huấn luyện (Continuous Learning): Flink đồng thời nạp toàn bộ sự kiện vào Apache Iceberg trên S3 (Offline Store) để định kỳ hàng ngày chạy Spark Job tái huấn luyện mô hình ML.

5. Góc Ôn Luyện Phỏng Vấn (Interview Corner)

❓ Câu hỏi 1: Giải thích cơ chế hoạt động của Checkpointing trong Apache Flink. Thuật toán Asynchronous Barrier Snapshotting (Chandy-Lamport) giúp Flink đạt được Exactly-Once State Consistency như thế nào mà không làm dừng luồng xử lý?

Gợi ý trả lời:
- Cơ chế hoạt động:
  - JobManager định kỳ bơm các điểm mốc đặc biệt gọi là Checkpoint Barriers vào dòng dữ liệu tại Source Operator.
  - Các Barrier này di chuyển đồng bộ cùng với dữ liệu qua các Operator trung gian (Transformation, Aggregation).
  - Khi một Operator nhận được Barrier từ tất cả các luồng đầu vào (Barrier Alignment), nó sẽ tạo một bản sao trạng thái nội bộ (State Snapshot) tại đúng mốc logic đó.
- Tính bất đồng bộ (Asynchronous):
  - Việc lưu trữ State thực tế ra đĩa ngoài (S3/HDFS) được thực hiện bởi một luồng ngầm (Background Thread).
  - Luồng chính của Flink tiếp tục xử lý các bản ghi tiếp theo ngay lập tức mà không cần đợi quá trình ghi đĩa hoàn tất, giúp loại bỏ hiện tượng nghẽn luồng (Stop-the-world).
- Khôi phục Exactly-Once:
  - Khi có sự cố sập node, tất cả Operator sẽ reset trạng thái về bản Checkpoint thành công gần nhất $N$ , và Kafka Consumer Offset sẽ được tua lại đúng mốc tương ứng với Barrier $N$ , đảm bảo không có bản ghi nào bị xử lý lặp lại vào State.

❓ Câu hỏi 2: Feature Store giải quyết vấn đề "Train-Serve Skew" như thế nào? Point-in-Time Correctness (hay Time-Travel Join) là gì?

Gợi ý trả lời:
- Train-Serve Skew (Lệch pha giữa Huấn luyện và Suy luận): Xảy ra khi công thức tính toán feature lúc train mô hình (viết bằng Python/SQL batch) khác biệt với công thức tính toán lúc chạy thực tế (viết bằng Java/Flink realtime), dẫn đến mô hình dự đoán sai lệch trên Production.
  - Giải pháp của Feature Store: Định nghĩa logic tính toán feature duy nhất một lần và dùng chung định nghĩa đó cho cả hai luồng Ingestion (Online và Offline).
- Point-in-Time Correctness (Tính đúng đắn theo mốc thời gian):
  - Khi huấn luyện mô hình dự đoán một giao dịch xảy ra lúc 2026-08-10 14:30:00, chúng ta cần các đặc trưng của người dùng tại đúng thời điểm đó (ví dụ: số dư tài khoản lúc 14:30:00).
  - Nếu dùng câu lệnh JOIN thông thường lấy giá trị hiện tại của bảng User, dữ liệu tương lai đã vô tình bị lộ cho mô hình (Data Leakage).
  - Feature Store thực hiện Time-Travel (ASOF) Join, đảm bảo mỗi dòng sự kiện lịch sử chỉ được ghép nối với giá trị feature có hiệu lực ngay trước thời điểm sự kiện diễn ra.

❓ Câu hỏi 3: Khi chạy một ứng dụng Stateful Streaming trên Flink với hàng Terabyte State, những kỹ thuật nào giúp bạn tối ưu hóa hiệu năng và tránh lỗi OutOfMemory (OOM)?

Gợi ý trả lời (4 kỹ thuật tối ưu cốt lõi):
1. Chuyển sang EmbeddedRocksDBStateBackend: Đưa toàn bộ State ra bộ nhớ đĩa cục bộ (SSD), tránh việc JVM Heap bị phình to gây tràn RAM và GC Pause.
2. Bật Incremental Checkpointing: Chỉ ghi các file SST mới sinh ra của RocksDB lên S3 thay vì ghi lại toàn bộ State, giúp giảm thời gian Checkpoint từ vài phút xuống vài giây và giảm 90% băng thông I/O.
3. Thiết lập State Time-To-Live (TTL): Tự động xóa bỏ các State cũ không còn giá trị (ví dụ: thông tin user không hoạt động quá 7 ngày) bằng cú pháp StateTtlConfig để kiểm soát dung lượng đĩa không tăng vô hạn.
4. Tối ưu hóa RocksDB Memory Allocator: Cấu hình state.backend.rocksdb.memory.managed = true để Flink tự động chia sẻ và kiểm soát vùng nhớ Off-heap giữa các RocksDB instances, tránh tình trạng tiến trình C++ native chiếm quá nhiều RAM làm hệ điều hành kích hoạt Linux OOM-Killer.

6. Tóm Tắt Ghi Nhớ Nhanh (Key Takeaways)

Apache Flink là sự lựa chọn số 1 cho các bài toán Stateful Streaming với độ trễ siêu thấp (Sub-second).
RocksDB State Backend + Incremental Checkpointing cho phép mở rộng State lên đến hàng chục Terabytes an toàn trên đĩa SSD.
Asynchronous Barrier Snapshotting (Chandy-Lamport) mang lại sự nhất quán Exactly-Once mà không làm gián đoạn luồng xử lý dữ liệu.
Feature Store là cấu trúc cầu nối giải quyết triệt để vấn đề Train-Serve Skew và đảm bảo Point-in-Time Correctness cho các hệ thống Real-Time AI/ML.