Posted inDevelopment

Hướng dẫn triển khai Change Data Capture (CDC) với Debezium và Apache Kafka: Đồng bộ dữ liệu thời gian thực từ PostgreSQL sang Elasticsearch

Bối cảnh: Vì sao Dual-write và Cronjob khiến bạn ôm hận?

Hình dung bạn đang vận hành một website thương mại điện tử với 2.000 đơn hàng mỗi phút. PostgreSQL giữ vai trò nguồn chân lý (Source of Truth), ghi nhận từng giao dịch chuẩn ACID. Cùng lúc đó, đội ngũ sản phẩm cần Elasticsearch để phục vụ người dùng tìm kiếm sản phẩm và lịch sử mua sắm với tốc độ phản hồi dưới 50ms.

Để giải bài toán này, giải pháp đầu tiên mọi người hay nghĩ tới là Dual-write: ứng dụng ghi vào Postgres xong thì gọi tiếp API bắn sang Elasticsearch. Nghe thì đơn giản. Nhưng chỉ cần Elasticsearch bị timeout 3 giây hoặc sập mạng giữa chừng, hai kho dữ liệu sẽ lệch nhau ngay lập tức. Bạn sẽ phải viết thêm hàng tá logic retry, bù trừ giao dịch cực kỳ rối rắm.

Lựa chọn thứ hai là dựng cronjob quét định kỳ mỗi 5 phút. Khi bảng dữ liệu cán mốc chục triệu dòng, câu lệnh SELECT ... WHERE updated_at > ... sẽ quét cạn IOPS của database. Đã vậy, dữ liệu tìm kiếm luôn trễ ít nhất 5 phút.

Đó là lý do kiến trúc Change Data Capture (CDC) trở thành cứu cánh. Thay vì ép database chạy câu lệnh nặng nề, CDC gắn một bộ đọc trực tiếp vào nhật ký giao dịch Write-Ahead Log (WAL) của PostgreSQL. Tác vụ thêm, sửa, xóa (INSERT, UPDATE, DELETE) đều được ghi nhận tức thì với chi phí CPU gần như bằng 0.

Bộ ba Debezium, Kafka và Kafka Connect xử lý trơn tru chuỗi mắt xích này: Debezium đọc WAL thành stream event, Kafka phân phối luồng tin cậy, còn Kafka Connect chuyển tiếp dữ liệu về Elasticsearch với độ trễ chỉ vài trăm mili-giây.

Khởi dựng môi trường với Docker Compose

Bạn không cần cài đặt rải rác từng công cụ lên máy cá nhân. Docker Compose sẽ giúp dựng trọn gói cụm dịch vụ này trong vài phút.

1. Chuẩn bị file docker-compose.yml

version: '3.8'
services:
  postgres:
    image: quay.io/debezium/postgres:15-alpine
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=inventory

  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kafka-connect:
    image: debezium/connect:2.4
    depends_on:
      - kafka
      - postgres
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: my_connect_configs
      OFFSET_STORAGE_TOPIC: my_connect_offsets
      STATUS_STORAGE_TOPIC: my_connect_statuses

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.10.2
    ports:
      - "9200:9200"
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false

2. Khởi động cụm dịch vụ

Chạy lệnh dưới nền:

docker compose up -d

Đợi khoảng 30 đến 45 giây để Kafka Connect nạp xong các plugin và mở cổng REST API 8083.

Cấu hình Pipeline CDC

Để Debezium trích xuất được log giao dịch, PostgreSQL bắt buộc phải kích hoạt wal_level = logical. Image quay.io/debezium/postgres đã bật sẵn cờ này. Nếu dùng instance Postgres có sẵn trên server hoặc RDS, bạn cần kiểm tra lại tham số này trong file cấu hình.

1. Tạo dữ liệu khởi tạo trong PostgreSQL

Tạo bảng khách hàng customers và chèn 2 bản ghi mẫu:

docker compose exec -it postgres psql -U postgres -d inventory -c "
CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    first_name VARCHAR(255),
    last_name VARCHAR(255),
    email VARCHAR(255) UNIQUE,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO customers (first_name, last_name, email) VALUES
('Nguyen', 'An', '[email protected]'),
('Tran', 'Binh', '[email protected]');
"

2. Đăng ký Debezium Postgres Source Connector

Gửi cấu hình connector qua REST API của Kafka Connect trên cổng 8083:

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ -d '{
  "name": "postgres-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "inventory",
    "topic.prefix": "cdc",
    "table.include.list": "public.customers"
  }
}'

Sau khi nhận lệnh, Debezium sẽ snapshot dữ liệu ban đầu. Toàn bộ thay đổi tiếp theo sẽ tự động đẩy vào topic Kafka mang tên cdc.public.customers.

3. Đẩy dữ liệu sang Elasticsearch với Unwrap SMT

Mặc định, message Debezium chứa phong bì (envelope) khá phức tạp gồm cấu trúc schema, trường before và after. Elasticsearch chỉ cần phần payload mới nhất. Ta dùng bộ biến đổi ExtractNewRecordState (Unwrap SMT) để bóc tách riêng object dữ liệu sạch trước khi index:

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ -d '{
  "name": "elastic-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "cdc.public.customers",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "_doc",
    "key.ignore": "false",
    "schema.ignore": "true",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false"
  }
}'

Nếu dự án không dùng plugin sink của Confluent, bạn có thể thay thế bằng một worker Python hay Go gọn nhẹ. Worker này đọc message từ Kafka và gọi _bulk API của Elasticsearch.

Kiểm thử & Vận hành thực tế

Hệ thống đã kết nối thông suốt. Giờ là lúc kiểm tra độ trễ thực tế.

1. Kiểm tra trạng thái Connector

Xác nhận cả hai connector đều đang chạy ổn định:

curl -s http://localhost:8083/connectors/postgres-source-connector/status | jq .
curl -s http://localhost:8083/connectors/elastic-sink-connector/status | jq .

Đảm bảo giá trị trường state hiển thị RUNNING. Nếu thấy FAILED, hãy xem log container bằng lệnh docker compose logs kafka-connect để bắt lỗi xác thực hoặc thiếu plugin.

2. Kiểm tra đồng bộ dữ liệu thời gian thực

Chèn thêm một khách hàng mới vào Postgres:

docker compose exec -it postgres psql -U postgres -d inventory -c "
INSERT INTO customers (first_name, last_name, email) VALUES ('Le', 'Cuong', '[email protected]');
"

Gần như tức thì (độ trễ thông thường dao động từ 150ms đến 300ms), truy vấn ngay trên Elasticsearch:

curl -s http://localhost:9200/cdc.public.customers/_search?q=last_name:Cuong | jq .hits.hits

Bản ghi của khách hàng ‘Le Cuong’ đã xuất hiện trong chỉ mục tìm kiếm mà ứng dụng không cần can thiệp bất kỳ dòng code đồng bộ nào.

3. Ba chỉ số sống còn khi đưa lên Production

Triển khai CDC trên môi trường thực tế đòi hỏi giám sát chặt chẽ các rủi ro sau:

  • Replication Slot Lag trên Postgres: Đây là bẫy phổ biến nhất. Nếu cụm Kafka Connect gặp sự cố dừng đột ngột, PostgreSQL vẫn giữ lại toàn bộ WAL file để chờ đọc tiếp. Ổ cứng server có thể phình từ 20GB lên 200GB chỉ sau vài giờ và làm tê liệt database chính. Luôn đặt alert theo dõi dung lượng WAL của replication slot qua câu lệnh pg_replication_slots.
  • Consumer Lag trên Kafka: Đo chênh lệch giữa offset mới nhất được tạo ra và offset mà Elasticsearch sink đã xử lý. Nếu chỉ số lag tăng liên tục, Elasticsearch đang bị nghẽn ghi. Bạn cần tăng thêm task hoặc tối ưu chỉ số refresh interval trên Elasticsearch.
  • Giám sát chỉ số JMX của Kafka Connect: Tích hợp Prometheus JMX Exporter để theo dõi tốc độ đọc bản ghi mỗi giây (poll rate) và thời gian cam kết offset (commit time), giúp bạn chủ động phát hiện nút thắt cổ chai trước khi khách hàng phàn nàn.
Share:
Tags: