Giới thiệu

Trong phần tiếp theo của series Debezium, chúng ta sẽ cùng nhau khám phá cách cấu hình Debezium kết hợp với PostgreSQL và Kafka Connect. Đây là một chủ đề quan trọng trong việc quản lý và truyền tải dữ liệu qua các hệ thống phân tán.

Debezium + Kafka Connect

Như đã đề cập trong các phần trước, Debezium hoạt động như một connector, giúp gửi các records vào Kafka. Trong khi đó, Kafka Connect đảm nhận vai trò phát hiện thay đổi dữ liệu và đẩy các sự kiện vào Kafka.

Thiết lập môi trường Debezium với Kafka Connect

Bài viết này sẽ sử dụng PostgreSQL làm ví dụ minh họa. Đầu tiên, bạn cần cài đặt PostgreSQL. Ngoài ra, chúng ta còn cần cài đặt một số thành phần sau:

1. Debezium UI: Giao diện web cho phép bạn cấu hình và quản lý các connector. Lưu ý rằng không có tính năng xác thực mặc định, nếu cần bảo mật, bạn có thể thiết lập NGINX bên ngoài.
2. Debezium Connect: Công cụ giúp theo dõi và nắm bắt các thay đổi ở phía cơ sở dữ liệu, đọc transaction log từ cơ sở dữ liệu gốc và gửi tới các dịch vụ streaming như Kafka, AWS Kinesis.
3. Kafka: Nền tảng streaming phân tán chủ yếu được sử dụng để vận chuyển thông điệp và thu thập, xử lý, lưu trữ dữ liệu.
4. Zookeeper: Dịch vụ phối hợp mở, phân tán dành cho các ứng dụng, thường đi kèm với Kafka.

Để dễ dàng triển khai, mình sẽ cung cấp một Dockerfile để bạn có thể làm demo một cách nhanh chóng. Dưới đây là nội dung của Dockerfile sử dụng phiên bản 1.7.2:

version: '3'
services:
  zookeeper:
    container_name: zookeeper
    image: quay.io/debezium/zookeeper:1.7.2.Final
    networks:
      - ui-network
    ports:
      - "2181:2181"
  kafka:
    container_name: kafka
    image: quay.io/debezium/kafka:1.7.2.Final
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
    environment:
      - ZOOKEEPER_CONNECT=zookeeper:2181
    networks:
      - ui-network
  db-pg:
    container_name: db-pg
    image: quay.io/debezium/example-postgres:1.7.2.Final
    ports:
      - "65432:5432"
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
    networks:
      - ui-network
  connect:
    container_name: connect
    image: quay.io/debezium/connect:nightly
    ports:
      - "8083:8083"
    depends_on:
      - kafka
      - db-pg
    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
      - ENABLE_DEBEZIUM_KC_REST_EXTENSION=true
      - ENABLE_DEBEZIUM_SCRIPTING=true
      - CONNECT_REST_EXTENSION_CLASSES=io.debezium.kcrestextension.DebeziumConnectRestExtension
    networks:
      - ui-network
  debezium-ui:
    container_name: debezium-ui
    image: quay.io/debezium/debezium-ui:1.7.2.Final
    ports:
      - "8080:8080"
    environment:
      - KAFKA_CONNECT_URIS=http://connect:8083
    depends_on:
      - connect
    networks:
      - ui-network
networks:
  ui-network:
    external: false

Sau khi chạy Docker, bạn có thể truy cập vào giao diện Web của Debezium UI tại localhost:8080. Nếu gặp vấn đề, hãy thử khởi động lại container connect vì nó có thể khởi động trước Kafka và gây ra lỗi.

Khi đã vào được giao diện, bạn có thể nhấn vào "Create a connector" để khám phá nhiều tính năng hấp dẫn. Để tiết kiệm thời gian, mình sẽ cung cấp cho bạn một ví dụ về cấu hình connector:

POST http://localhost:8080/api/connector/1/postgres
{
    "name": "test.31",
    "config": {
        "topic.prefix": "prefix",
        "database.hostname": "db-pg",
        "database.port":"5432",
        "database.user": "postgres",
        "database.password": "postgres",
        "database.dbname": "postgres",
        "schema.include.list": "inventory",
        "table.include.list": "inventory.orders ",
        "heartbeat.interval.ms": 30000,
        "heartbeat.action.query": "select * from inventory.orders where id = 10001",
        "query.fetch.size": 500,
        "topic.creation.groups": "debezium-etl",
        "topic.creation.debezium-etl.include": "",
        "topic.creation.debezium-etl.exclude": "",
        "topic.creation.default.partitions": -1,
        "topic.creation.default.replication.factor": -1,
        "plugin.name": "pgoutput",
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "tasks.max": "1",
        "skipped.operations": "r",
        "snapshot.mode": "never",
        "slot.name": "debezium_order",
        "publication.autocreate.mode": "filtered",
        "publication.name": "dbz_order_publication"
    }
}

Nếu bạn không rõ ý nghĩa của các cấu hình này, hãy đọc lại phần 2 của series để hiểu rõ hơn.

Giới thiệu về Publication

Trong Debezium, bạn có thể chọn ba loại publication để cấu hình:

1. all_tables: Nếu một publication tồn tại, connector sẽ sử dụng. Nếu không, nó sẽ tự động tạo một publication cho tất cả các bảng.
2. disabled: Connector không cố gắng tạo ra một publication mà cần quản trị viên tạo trước.
3. filtered: Tương tự như all_tables nhưng cho phép lọc các bảng thông qua cấu hình chỉ định.

Thông thường, việc sử dụng filtered là lựa chọn tốt hơn vì nó yêu cầu ít quyền hơn.

Kiểm tra kết quả

Sau khi có connector hoạt động, bạn có thể thực hiện các câu lệnh SQL sau để kiểm tra:

SELECT slot_name, plugin, slot_type FROM pg_replication_slots;

SELECT * FROM pg_stat_replication;

SELECT * FROM pg_publication;

SELECT * FROM pg_publication_tables WHERE pubname = 'dbz_order_publication';

Hai câu SQL đầu tiên sẽ phản ánh slot replication và trạng thái kết nối đến connector. Hai câu sau sẽ cung cấp thông tin chi tiết về publication được tạo ra. Với việc này, bạn có thể kết nối đến Kafka và xử lý dữ liệu theo nhu cầu của mình.

Trên đây là phần hướng dẫn cơ bản về việc cấu hình Debezium với PostgreSQL và Kafka Connect. Nếu bạn gặp phải bất kỳ vấn đề nào trong suốt quá trình thiết lập, hãy liên hệ với mình, mình sẽ cố gắng hỗ trợ bạn. Chúc bạn thành công với những kiến thức mới này!

Series

• Phần 1: Debezium là gì? Ứng dụng thực tế.
• Phần 2: Cấu hình sử dụng Debezium Engine + PostgreSQL.
• Phần 3: Cấu hình sử dụng Debezium + PostgreSQL + Kafka Connect.
  source: viblo