Giới thiệu về cơ chế lưu trữ trạng thái

Trong quy trình tích hợp Flink CDC với PostgreSQL, việc quản lý trạng thái (state) và cơ chế checkpoint đóng vai trò then chốt để đảm bảo tính chính xác và khả năng phục hồi của luồng dữ liệu. Flink cung cấp nhiều tùy chọn StateBackend khác nhau, mỗi loại phù hợp với các yêu cầu về hiệu năng và độ bền vững cụ thể.

Các loại StateBackend hỗ trợ

Dưới đây là phân tích chi tiết về ba loại StateBackend chính thường được sử dụng trong các作业 Flink:

1. MemoryStateBackend

• Đặc điểm: Dữ liệu trạng thái được lưu trữ trên heap JVM của TaskManager, trong khi dữ liệu checkpoint được giữ lại trong bộ nhớ của JobManager.
• Trường hợp sử dụng: Phù hợp cho môi trường phát triển, kiểm thử đơn vị hoặc các作业 có trạng thái nhỏ (ví dụ: cửa sổ thời gian ngắn).
• Hạn chế: Kích thước trạng thái bị giới hạn bởi bộ nhớ TaskManager và không hỗ trợ cơ chế High Availability (HA).

2. FsStateBackend

• Đặc điểm: Trạng thái nằm trên heap của TaskManager, nhưng các checkpoint được ghi vào hệ thống tệp tin phân tán (HDFS, S3) hoặc hệ thống tệp cục bộ.
• Trường hợp sử dụng: Thích hợp cho các作业 sản xuất có trạng thái trung bình, yêu cầu khả năng phục hồi sau sự cố.
• Lợi ích: Hỗ trợ snapshot bất đồng bộ, giảm thiểu độ trễ trong quá trình xử lý luồng.

3. RocksDBStateBackend

• Đặc điểm: Sử dụng RocksDB nhúng để lưu trữ trạng thái trên disk cục bộ của TaskManager, checkpoint được đẩy lên hệ thống tệp tin bên ngoài.
• Trường hợp sử dụng: Bắt buộc cho các作业 có trạng thái rất lớn (cửa sổ dài, key-value lớn) hoặc yêu cầu HA nghiêm ngặt.
• Lợi ích: Hỗ trợ checkpoint gia tăng (incremental), tối ưu hóa băng thông mạng và dung lượng lưu trữ.

Hướng dẫn triển khai mã nguồn

Các ví dụ dưới đây minh họa cách cấu hình từng loại StateBackend trong ứng dụng Java sử dụng Flink CDC.

1. Thiết lập MemoryStateBackend

Mặc định Flink đã sử dụng cấu hình này. Tuy nhiên, để chủ động kiểm soát, bạn có thể khởi tạo rõ ràng như sau:

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;

public class PostgresCdcJobConfig {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Khởi tạo bộ nhớ trạng thái trên JVM
        streamEnv.setStateBackend(new MemoryStateBackend());
        
        // Kích hoạt checkpoint mỗi 6 giây
        streamEnv.enableCheckpointing(6000);
        
        // Cấu hình nguồn CDC PostgreSQL (giả định)
        // PostgresSource<String> cdcSource = PostgresSource.<String>builder()...build();
        // streamEnv.addSource(cdcSource).print();
        
        streamEnv.execute("Postgres CDC - Memory Backend");
    }
}

2. Thiết lập FsStateBackend

Cấu hình này yêu cầu chỉ định đường dẫn lưu trữ checkpoint bên ngoài:

import org.apache.flink.runtime.state.filesystem.FsStateBackend;

public class PostgresCdcJobConfig {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Trỏ checkpoint tới hệ thống tệp tin cục bộ hoặc HDFS
        streamEnv.setStateBackend(new FsStateBackend("file:///var/data/flink/checkpoints"));
        
        // Cấu hình thời gian checkpoint và timeout
        streamEnv.enableCheckpointing(15000);
        streamEnv.getCheckpointConfig().setCheckpointTimeout(120000);
        
        // Thêm nguồn dữ liệu CDC
        // streamEnv.addSource(cdcSource).print();
        
        streamEnv.execute("Postgres CDC - Fs Backend");
    }
}

3. Thiết lập RocksDBStateBackend

Để sử dụng RocksDB, cần đảm bảo thư viện tương thích đã được thêm vào dự án.

Thêm phụ thuộc Maven:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-rocksdb_2.12</artifactId>
    <version>1.13.6</version>
</dependency>

Cấu hình trong code:

import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;

public class PostgresCdcJobConfig {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Bật chế độ incremental checkpoint cho RocksDB
        streamEnv.setStateBackend(new RocksDBStateBackend("file:///var/data/flink/checkpoints", true));
        
        // Tối ưu hóa tần suất checkpoint
        streamEnv.enableCheckpointing(45000);
        streamEnv.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
        streamEnv.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        
        // Thực thi job
        streamEnv.execute("Postgres CDC - RocksDB Backend");
    }
}

Các tham số cấu hình quan trọng

• Đường dẫn lưu trữ: Đối với FsStateBackend và RocksDBStateBackend, đường dẫn checkpoint phải trỏ tới hệ thống tệp tin chia sẻ (HDFS, S3) hoặc disk có quyền ghi. Ví dụ: hdfs://namenode:8020/flink/checkpoints.
• Checkpoint gia tăng: Chỉ có trên RocksDBStateBackend, giúp giảm đáng kể thời gian và dung lượng lưu trữ mỗi lần snapshot bằng cách chỉ lưu phần thay đổi.
• High Availability (HA): Để đảm bảo tính sẵn sàng cao, cần cấu hình state.checkpoints.dir và state.savepoints.dir trỏ tới storage chung, kết hợp với Zookeeper hoặc Kubernetes HA.

Chiến lược lựa chọn StateBackend

• Môi trường Dev/Test: Ưu tiên MemoryStateBackend để đơn giản hóa việc triển khai và debug.
• Sản xuất (Trạng thái nhỏ): Sử dụng FsStateBackend để cân bằng giữa tốc độ truy xuất bộ nhớ và độ bền vững của dữ liệu.
• Sản xuất (Trạng thái lớn/HA): Bắt buộc dùng RocksDBStateBackend để xử lý khối lượng dữ liệu trạng thái vượt quá RAM và hỗ trợ khôi phục sự cố hiệu quả.