Posted inDatabase

Change Data Capture với Debezium và Apache Kafka: Truyền Dữ liệu Thay đổi từ PostgreSQL theo Thời gian Thực

Database tutorial - IT technology blog
Database tutorial - IT technology blog

Vấn đề Thực tế: Giữ Đồng bộ Các Hệ thống Phía Sau (Downstream)

Các cơ sở dữ liệu trên môi trường production hiếm khi hoạt động độc lập. Hãy xem xét một quy trình thanh toán tiêu chuẩn: một cú nhấp chuột duy nhất của khách hàng sẽ ghi một dòng đơn hàng vào PostgreSQL, làm mất hiệu lực session trong cache Redis, đánh lại chỉ mục catalog trên Elasticsearch và kích hoạt webhook gửi tới Stripe.

Các nhóm phát triển thường bắt đầu với hai mô hình đơn giản: ghi đồng thời hai nơi (dual-write) trực tiếp trong mã nguồn ứng dụng hoặc chạy các truy vấn cron định kỳ như SELECT * FROM orders WHERE updated_at > :last_poll_time mỗi 30 giây một lần. Mặc dù cách này hoạt động ổn trong giai đoạn thử nghiệm ban đầu, nhưng nó nhanh chóng bộc lộ vấn đề khi lưu lượng truy cập tăng lên hàng trăm lượt ghi mỗi giây:

  • Lệch Dữ liệu Âm thầm (Silent Data Drift): Nếu worker của ứng dụng bị crash ngay sau khi commit vào PostgreSQL nhưng trước khi kịp gọi Elasticsearch, search index của bạn sẽ bị lỗi thời. Nếu không có các script đối soát thủ công, sự sai lệch này sẽ tồn tại vô thời hạn.
  • Nghẽn I/O Cơ sở Dữ liệu: Việc polling định kỳ mỗi 5 đến 10 giây buộc PostgreSQL phải quét liên tục các phạm vi index lớn. Với bảng dữ liệu 10 triệu dòng, các truy vấn đó sẽ ngốn sạch băng thông đĩa và chu kỳ CPU mà các tác vụ giao dịch cốt lõi đang cần.
  • Bỏ sót Thao tác Xóa Hoàn toàn (Hard Deletes): Một truy vấn polling lọc theo trường updated_at sẽ không bao giờ phát hiện được các bản ghi bị xóa hoàn toàn (DELETE FROM orders WHERE id = 42). Khi đó, bạn buộc phải dùng đến giải pháp xóa mềm (soft-delete) cồng kềnh trên mọi bảng hoặc phải xây dựng các bảng riêng để theo dõi việc dọn dẹp dữ liệu.

Phân tích Nguyên nhân Gốc rễ: Vì sao Các Chiến lược Đồng bộ Truyền thống Thất bại

Những sự cố đồng bộ này bắt nguồn từ hai hạn chế về mặt kiến trúc: sự thiếu vắng các giao dịch nguyên tử phân tán (distributed atomic transactions) và việc bỏ qua hoàn toàn nhật ký nội bộ (engine logs) của PostgreSQL.

1. Race Condition khi Ghi Đồng thời (Dual-Write)

Ghi dữ liệu vào cơ sở dữ liệu chính và đẩy vào hàng đợi phụ trong cùng một request handler sẽ tạo ra nguy cơ phân mảnh (split-brain) không thể tránh khỏi. Nếu mạng gặp sự cố giữa lúc commit cơ sở dữ liệu và publish vào hàng đợi, sự kiện đó sẽ bị mất vĩnh viễn. Nếu bạn retry một cách ngây thơ, bạn sẽ đối mặt với rủi ro thông điệp bị trùng lặp hoặc sai thứ tự. Về lý thuyết, giao thức Two-phase commit (2PC) có thể điều phối được, nhưng chúng lại làm tăng độ trễ rất lớn (thường là 50–100ms cho mỗi giao dịch) và rất dễ thất bại bất thường khi xảy ra phân vùng mạng (network partition).

2. Polling Bỏ qua Các Trạng thái Thay đổi Trung gian

Hãy hình dung một đơn hàng chuyển từ trạng thái PENDING sang PROCESSING rồi sang SHIPPED chỉ trong vòng 2 giây. Nếu truy vấn polling của bạn chạy 10 giây một lần, trạng thái trung gian PROCESSING sẽ biến mất hoàn toàn. Các pipeline phân tích và dịch vụ phát hiện gian lận sẽ bỏ lỡ các bước chuyển đổi trạng thái quan trọng này, dẫn đến việc bạn chỉ thu thập được lịch sử kiểm toán (audit) không trọn vẹn.

So sánh Các Hướng Tiếp cận Đồng bộ Dữ liệu

Dưới đây là bảng so sánh 4 mô hình truyền bá thay đổi phổ biến nhất trong môi trường production:

Hướng tiếp cận Độ trễ Tải lên Cơ sở Dữ liệu Xử lý Lệnh Xóa? Độ tin cậy
Application Dual-Writes Thấp (<50ms) Thấp Có Kém (Dễ lệch dữ liệu & lỗi cục bộ)
Scheduled Polling (ETL) Cao (10s – Hàng giờ) Cao (Chiếm dụng CPU & Disk I/O liên tục) Không (Yêu cầu soft deletes) Trung bình (Bỏ sót các trạng thái trung gian)
Bảng Dựa trên Trigger Thấp (<100ms) Cao (Khuếch đại ghi gấp 2 lần) Có Tốt (Bảo trì schema phức tạp)
Log-Based CDC (Debezium + Kafka) Dưới 1 giây (<10ms) Gần như bằng 0 (Đọc tuần tự từ WAL) Có Tuyệt vời (Đảm bảo thứ tự nghiêm ngặt & tính nguyên tử)

Log-based Change Data Capture (CDC) hoạt động bằng cách bỏ qua hoàn toàn tầng SQL. PostgreSQL ghi nhận tuần tự mọi thao tác insert, update và commit vào Write-Ahead Log (WAL) trước khi ghi dữ liệu xuống đĩa. Các engine CDC đọc trực tiếp nhật ký này. Nhờ đó, bạn nhận được thông báo thay đổi với độ trễ chỉ tính bằng mili-giây, không gây tải truy vấn lên dữ liệu bảng đang hoạt động, đảm bảo thứ tự giao dịch nghiêm ngặt và độ chính xác 100% trên từng thao tác dòng.

Khi khởi tạo dữ liệu thử nghiệm hoặc chuẩn bị file seed ban đầu cho các consumer phía sau, sự bất đồng bộ về định dạng thường xuyên xảy ra. Khi cần chuyển đổi nhanh CSV sang JSON để nhập dữ liệu, tôi thường dùng toolcraft.app/vi/tools/data/csv-to-json — công cụ này chạy trực tiếp trên trình duyệt nên dữ liệu không hề rời khỏi máy của bạn. Khi cấu trúc schema chuẩn đã sẵn sàng, chúng ta có thể tiến hành thiết lập pipeline thời gian thực.

Triển khai Log-Based CDC với Debezium và Kafka

Debezium kết nối với PostgreSQL như một replication client tiêu chuẩn thông qua plugin giải mã logic tích hợp sẵn pgoutput. Nó đọc các thay đổi nhị phân từ WAL, giải tuần tự hóa (deserialize) các giá trị dòng thành các sự kiện có cấu trúc và gửi chúng vào các topic riêng biệt trong Apache Kafka.

Bước 1: Cấu hình PostgreSQL cho Logical Replication

PostgreSQL cần ghi chi tiết các thay đổi ở cấp độ hàng vào các file WAL. Hãy thiết lập các tham số sau trong file postgresql.conf của bạn:

wal_level = logical
max_wal_senders = 4
max_replication_slots = 4

Tạo một người dùng chuyên dụng cho replication và một bảng mẫu:

CREATE USER cdc_user WITH REPLICATION PASSWORD 'secret_cdc_pass';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_user;

CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Ghi lại toàn bộ trạng thái hàng trước khi thay đổi trong các thao tác UPDATE và DELETE
ALTER TABLE customers REPLICA IDENTITY FULL;

Bước 2: Khởi chạy Hệ thống với Docker Compose

Dưới đây là file docker-compose.yml hoàn chỉnh chạy PostgreSQL 15, Apache Kafka (chế độ KRaft không cần ZooKeeper) và Debezium Kafka Connect:

version: '3.8'
services:
  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: appdb
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: rootpassword
    command: ["postgres", "-c", "wal_level=logical"]
    ports:
      - "5432:5432"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
      KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9092'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
    ports:
      - "9092:9092"

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

Bước 3: Đăng ký Debezium PostgreSQL Connector

Gửi cấu hình connector tới REST interface của Kafka Connect tại cổng 8083:

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ -d '{
  "name": "postgres-cdc-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": "rootpassword",
    "database.dbname": "appdb",
    "topic.prefix": "pg_cdc",
    "table.include.list": "public.customers"
  }
}'

Bước 4: Kiểm tra Luồng Dữ liệu Thay đổi

Thêm một dòng mới vào PostgreSQL:

INSERT INTO customers (name, email) VALUES ('Alex Rivers', '[email protected]');

Đọc trực tiếp message được tạo ra từ Kafka topic pg_cdc.public.customers:

docker exec -it $(docker ps -qf "name=kafka") kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic pg_cdc.public.customers \
  --from-beginning

Debezium sẽ xuất ra một envelope toàn diện chi tiết các trạng thái trước và sau khi thay đổi cùng với metadata của giao dịch:

{
  "before": null,
  "after": {
    "id": 1,
    "name": "Alex Rivers",
    "email": "[email protected]",
    "updated_at": 1700000000000
  },
  "source": {
    "version": "2.4.0.Final",
    "connector": "postgresql",
    "name": "pg_cdc",
    "ts_ms": 1700000000123,
    "db": "appdb",
    "schema": "public",
    "table": "customers",
    "txId": 582
  },
  "op": "c",
  "ts_ms": 1700000000456
}

Trường op xác định rõ ràng loại sự kiện: c là tạo mới (insert), u là cập nhật (update), d là xóa (delete), và r là lần đọc snapshot ban đầu. Các consumer phía sau có thể đảm bảo tính bất biến (idempotency) bằng cách dùng id làm khóa và kiểm tra source.txId.

Những Lưu ý Quan trọng trên Môi trường Production

  • Phình to Replication Slot: Nếu Debezium bị ngắt kết nối nhiều giờ liền trong khi lưu lượng ghi vẫn tiếp diễn (ví dụ: 2.000 lượt ghi/giây), PostgreSQL sẽ giữ lại toàn bộ các log WAL chưa được tiêu thụ trên ổ đĩa. Hãy luôn cấu hình max_slot_wal_keep_size (ví dụ: 10GB) và giám sát pg_replication_slots.active để tránh sự cố đầy đĩa làm sập hệ thống.
  • Sự kiện Tombstone: Sau một thao tác xóa, Debezium sẽ phát ra một bản ghi có payload mang giá trị null. Điều này cho phép log cleaner của Kafka dọn dẹp các khóa bản ghi cũ trong quá trình compaction. Hãy đảm bảo consumer phía sau của bạn xử lý tốt các payload null mà không bị lỗi NullPointerException.
  • Migration Schema: Việc thêm một cột chấp nhận giá trị null (nullable column) thông qua ALTER TABLE sẽ được truyền đi một cách mượt mà mà không cần khởi động lại connector. Tuy nhiên, các consumer phía sau cần parse JSON payload một cách cẩn trọng để thích ứng linh hoạt với sự thay đổi của schema.
Share:
Tags: