Nếu các bạn chưa biết Debezium là gì thì có thể theo dõi bài viết này trước: https://viblo.asia/p/gioi-thieu-ve-debezium-y3RL1wMp4ao
Công cụ cần thiết:
- Docker
- Docker compose
Các bước cài đặt
1. Tạo file docker-compose.yml
📌Tạo một thư mục, vào thư mục đó và tạo file docker-compose.yml:
version: '3.8'
services: zookeeper: image: confluentinc/cp-zookeeper:latest container_name: zookeeper environment: ZOOKEEPER_CLIENT_PORT: 2181 ports: - "2181:2181" kafka: image: confluentinc/cp-kafka:latest container_name: kafka depends_on: - zookeeper environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 ports: - "9092:9092" postgres: image: debezium/postgres:16-alpine container_name: postgres environment: POSTGRES_USER: debezium POSTGRES_PASSWORD: debezium POSTGRES_DB: testdb ports: - "5432:5432" debezium-connect: image: debezium/connect:2.5 # Sửa lại từ "latest" thành phiên bản cụ thể container_name: debezium-connect depends_on: - kafka - postgres environment: BOOTSTRAP_SERVERS: kafka:9092 GROUP_ID: "1" CONFIG_STORAGE_TOPIC: debezium_connect_configs OFFSET_STORAGE_TOPIC: debezium_connect_offsets STATUS_STORAGE_TOPIC: debezium_connect_status ports: - "8083:8083" kafka-ui: image: provectuslabs/kafka-ui:latest container_name: kafka-ui depends_on: - kafka environment: KAFKA_CLUSTERS_0_NAME: local KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092 ports: - "8080:8080" 📌Lưu file này lại và chạy lệnh sau để khởi động Docker containers:
docker-compose up -d
2. Tạo table muốn CDC trong PostgreSQL
📌 Sau khi khởi động thành công, chúng ta sẽ vào PostgreSQL để tạo table . Ví dụ
CREATE TABLE orders ( id SERIAL PRIMARY KEY, product_name TEXT NOT NULL, status TEXT NOT NULL
);
📌 Kích hoạt Logical Replication cho PostgreSQL
Debezium sử dụng Logical Replication để theo dõi các thay đổi trong database. Do đó, chúng ta cần:
Bật chế độ logical replication trong PostgreSQL
ALTER SYSTEM SET wal_level = logical; #Cho phép PostgreSQL ghi lại dữ liệu thay đổi (change stream) dưới dạng logical, thay vì physical.
SELECT pg_reload_conf(); #Load lại cấu hình mà không cần restart database.
📌 Tạo publication cho các bảng cần theo dõi:
Tạo publication để chỉ định bảng nào cần theo dõi
CREATE PUBLICATION debezium_pub FOR TABLE orders; #Chỉ định rằng chỉ cần theo dõi bảng orders
3. Cấu hình Debezium để theo dõi PostgreSQL
Mở Gitbash lên và chạy curl :
curl -X POST "http://localhost:8083/connectors" \ -H "Content-Type: application/json" \ -d '{ "name": "postgres-connector", "config": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "database.hostname": "postgres", "database.port": "5432", "database.user": "debezium", "database.password": "debezium", "database.dbname": "testdb", "database.server.name": "postgres_server", "table.include.list": "public.orders", "plugin.name": "pgoutput", "slot.name": "debezium_slot", "publication.name": "debezium_pub", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "schema-changes", "topic.prefix": "dbserver1" #tạo topic với prefix là dbserver1 } }' 4. Thử nghiệm thêm dữ liệu
📌 Mở terminal và truy cập PostgreSQL:
docker exec -it postgres psql -U debezium -d testdb
insert vào bảng order
INSERT INTO orders (product_name, status) VALUES ('Laptop', 'PENDING');
5. Kiểm tra dữ liệu trên Kafka
📌 Mở Kafka UI tại 👉 http://localhost:8080
📌 Chọn topic postgresserver.public.orders để xem dữ liệu.
Đây là topic mà debezium publish message lên :
Và đây là thành quả :
Các bạn cũng có thể custom lại được message này trước khi publish message lên topic.
Tớ sẽ hướng dẫn mọi người cách custom message ở bài sau.
Nếu các bạn có bất kỳ thắc mắc nào hãy comment ở bên dưới mình sẽ giải đáp. 😁😁😁😁