PostgreSQL CDC c помощью Debezium и Kafka (Часть 1)

Про саму технологию CDC отлично написано в статье.

В данной публикации мы развернем Debezium и Kafka в докере, настроим подключение и получим наши первые данные.

Немного теории:

Debezium — это платформа для захвата изменений в базе данных (CDC, Change Data Capture), которая позволяет отслеживать все изменения в данных и передавать их в другие системы в реальном времени.

В связке с PostgreSQL и Kafka Debezium работает следующим образом:

1. PostgreSQL: Лог репликации

Debezium использует лог репликации PostgreSQL, чтобы отслеживать изменения в базе данных. PostgreSQL поддерживает механизмы логирования всех операций с данными (INSERT, UPDATE, DELETE) в виде так называемых WAL (Write-Ahead Logs). Эти логи содержат последовательность изменений, которые происходят в базе данных.

Когда мы включаем в PostgreSQL "wal_level = logical " и подключаем Debezium, то он читает эти логи и отслеживает изменения. Debezium не вносит изменения в базу данных , а просто читает, что было добавлено, изменено или удалено.

2. Debezium: Коннектор для PostgreSQL

Debezium предоставляет коннектор для PostgreSQL, который подключается к этой базе данных и начинает читать WAL. Коннектор работает как агент, который «слушает» изменения. При каждом изменении данных (например, новая запись, обновление или удаление) Debezium преобразует это изменение в событие.

3. Kafka: Передача событий

После того как Debezium считывает изменения, оно передает эти события в Kafka — систему передачи сообщений. А Kafka собирает, сохраняет и передает данные между различными системами, не только 1С.

Debezium отправляет события изменения данных в виде сообщений в определенные топики Kafka. Например, если мы изменяем запись в таблице с Номенклатурой, то событие с информацией об этом изменении будет отправлено в топик, который может называться что-то вроде buhbase.nomenclature. Каждое событие может содержать информацию о том:

• Что это было за изменение (вставка, обновление, удаление).
• Какие данные изменились (старые и новые значения).

Общий поток выглядит так:

1. PostgreSQL пишет изменения в WAL.
2. Debezium читает эти изменения и преобразует их в события.
3. Kafka получает эти события в виде сообщений и распределяет их по топикам.
4. Потребители Kafka могут использовать эти события для разных нужд, как, например, обновление данных, обработка или репликация.

Таким образом, Debezium + PostgreSQL + Kafka позволяет в реальном времени отслеживать все изменения в базе данных и использовать эти данные для различных интеграций и потоковой обработки.

Для начала работы у вас должен быть установлен Docker.

Содержимое файла Docker compose:

version: '5.0'
services:

  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    hostname: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 2181:2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    hostname: kafka
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kconnect:
    image: debezium/connect:latest
    ports:
      - 8083:8083
    environment:
      CONFIG_STORAGE_TOPIC: my_connect_configs
      OFFSET_STORAGE_TOPIC: my_connect_offsets
      STATUS_STORAGE_TOPIC: my_connect_statuses
      BOOTSTRAP_SERVERS: kafka:29092   
    links:
      - zookeeper
    depends_on:
      - kafka
      - zookeeper

Сохраним содержимое в файл docker-compose.yml.

В командной строке перейдем в папку с файлом и запустим команду:  docker-compose up -d

Должны увидеть подобную картину:

А в приложении Doker запущенные наши контейнеры:

Для нормального соединения к контейнерам по имени, я еще пропишу в файл C:\Windows\System32\drivers\etc\hosts соответствие имении и IP адреса.

С помощью программы Offset explorer подключимся к Kafka и проверим, что брокер работает.

Более того, мы видим топики, созданные Debezium.

Теперь настроим связь между Debezium и PostgreSQL. В базе 1С у меня есть таблица _reference47, будем подключать её.

Управление Debezium производится с помощью REST - интерфейса, далее я буду использовать Postman.

Запустим наш запрос, после этого мы можем проверить, что коннектор добавился с помощью GET запроса.

Добавим информацию в нашем справочнике и посмотрим, что у нас появилось в Kafka.

Таким образом мы настроили самый базовый сценарий CDC.