Поставка данных из Managed Service for PostgreSQL в Managed Service for Apache Kafka® с помощью Debezium

Вы можете отслеживать изменения данных в Managed Service for PostgreSQL и отправлять их в Managed Service for Apache Kafka® с помощью технологии Change Data Capture (CDC).

Далее будет продемонстрировано на примере, как настроить CDC с помощью Debezium.

Перед началом работы

1. Создайте кластер-источник Managed Service for PostgreSQL:

   • с хостами в публичном доступе;
   • с базой данных db1;
   • с пользователем user1.

2. Создайте кластер-приемник Managed Service for Apache Kafka® любой подходящей вам конфигурации с хостами в публичном доступе.

3. Создайте виртуальную машину для Debezium с Ubuntu 20.04 и публичным IP-адресом.

4. Настройте группы безопасности так, чтобы к кластерам можно было подключиться из созданной виртуальной машины:

   • Managed Service for PostgreSQL;
   • Managed Service for Apache Kafka®.

5. Подключитесь к виртуальной машине по SSH и установите зависимости:

   sudo apt update && \
   sudo apt install -y git docker.io kafkacat postgresql-client
   

6. Подключитесь к виртуальной машине по SSH и проверьте, что доступны кластеры:

   • Managed Service for PostgreSQL (используйте утилиту psql);
   • Managed Service for Apache Kafka® (используйте утилиту kafkacat).

Подготовка кластера-источника

Настройте Debezium

1. Подключитесь к виртуальной машине по SSH.

2. Склонируйте репозиторий:

   cd ~/ && \
   git clone https://github.com/yandex-cloud/examples.git
   

3. Скачайте и распакуйте актуальный Debezium-коннектор в директорию ~/examples/mdb/managed-kafka/debezium-cdc/plugins/.

   Ниже приведен пример для версии 1.6.0. Нужные команды зависят от типа кластера-источника:

   Managed Service for PostgreSQL

   mkdir  ~/examples/mdb/managed-kafka/debezium-cdc/plugins/ && \
   wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.6.0.Final/debezium-connector-postgres-1.6.0.Final-plugin.tar.gz && \
   tar -xzvf debezium-connector-postgres-1.6.0.Final-plugin.tar.gz -C ~/examples/mdb/managed-kafka/debezium-cdc/plugins/
   

4. Соберите образ Docker:

   cd ~/examples/mdb/managed-kafka/debezium-cdc && \
   sudo docker build --tag debezium ./
   

5. Создайте файл ~/examples/mdb/managed-kafka/debezium-cdc/mdb-connector.properties с настройками Debezium для подключения к кластеру-источнику:

   Managed Service for PostgreSQL

   name=debezium-mpg
   connector.class=io.debezium.connector.postgresql.PostgresConnector
   plugin.name=pgoutput
   database.hostname=c-<идентификатор кластера>.rw.mdb.yandexcloud.net
   database.port=6432
   database.user=user1
   database.password=<пароль пользователя user1>
   database.dbname=db1
   database.server.name=mpg
   table.include.list=public.measurements
   publication.name=mpg_publication
   slot.name=debezium_slot
   heartbeat.interval.ms=15000
   heartbeat.topics.prefix=__debezium-heartbeat
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге.

   Здесь:

   • name — имя коннектора Debezium.
   • database.hostname — особый FQDN для подключения к хосту-мастеру кластера-источника.
   • database.user — имя пользователя PostgreSQL.
   • database.dbname — имя базы данных PostgreSQL.
   • database.server.name — произвольное имя сервера баз данных, которое Debezium будет использовать при выборе топика для отправки сообщений.
   • table.include.list — список имен таблиц, для которых Debezium должен отслеживать изменения. Укажите полные имена, включающие в себя имя схемы (по умолчанию public). Debezium будет использовать значения настроек из этого поля при выборе топика для отправки сообщений.
   • publication.name — имя публикации, созданной на кластере-источнике.
   • slot.name — имя слота репликации, который будет создан Debezium при работе с публикацией.
   • heartbeat.interval.ms и heartbeat.topics.prefix — настройки heartbeat, необходимые для работы Debezium.

Подготовьте кластер-приемник

Настройки кластера-приемника зависят от типа кластера-источника:

Запустите Debezium

1. Укажите в файле ~/examples/mdb/managed-kafka/debezium-cdc/.env переменные окружения, используемые для доступа к кластеру-приемнику:

   BROKERS=<FQDN хоста-брокера>:9091
   USER=debezium
   PASSWORD=<пароль пользователя debezium>
   

   Список FQDN хостов-брокеров можно запросить со списком хостов в кластере.

2. Запустите контейнер с собранным Docker-образом:

   cd ~/examples/mdb/managed-kafka/debezium-cdc/ && \
   sudo docker run --name debezium --rm --env-file .env \
     -v ~/examples/mdb/managed-kafka/debezium-cdc/plugins:/home/appuser/plugins \
     -v ~/examples/mdb/managed-kafka/debezium-cdc/mdb-connector.properties:/home/appuser/config/connector.properties \
     debezium:latest
   

   Контейнер будет непрерывно передавать новые данные из кластера-источника в кластер-приемник.

Проверьте работоспособность Debezium

1. Подключитесь к кластеру-приемнику, например, с помощью kafkacat, и проверьте, что данные поступают.

   В сообщениях топика mpg.public.measurements должны присутствовать данные и схема их формата.

   Пример фрагмента сообщения

   {
   "schema": {
       ...
   },
   "payload": {
       "before": null,
       "after": {
           "device_id": "iv9a94th6rztooxh5ur2",
           "datetime": 1591378020000000,
           "latitude": 55.70329,
           "longitude": 37.65472,
           "altitude": 427.5,
           "speed": 0.0,
           "battery_voltage": 23.5,
           "cabin_temperature": 17.0,
           "fuel_level": null
       },
       "source": {
           "version": "1.6.0.Final",
           "connector": "postgresql",
           "name": "mpg",
           "ts_ms": 1628245046882,
           "snapshot": "true",
           "db": "db1",
           "sequence": "[null,\"4328525512\"]",
           "schema": "public",
           "table": "measurements",
           "txId": 8861,
           "lsn": 4328525328,
           "xmin": null
       },
       "op": "r",
       "ts_ms": 1628245046893,
       "transaction": null
     }
   }
   

2. Добавьте данные в таблицу на кластере-источнике:

   Managed Service for PostgreSQL

   1. Подключитесь к кластеру.

   2. Добавьте еще одну строку в таблицу measurements:

      INSERT INTO public.measurements VALUES ('iv7b74th678tooxh5ur2', '2020-06-08 17:45:00', 53.70987913, 36.62549834, 378.0, 20.5, 5.3, 20, NULL);
      

3. Подключитесь к кластеру-приемнику и убедитесь, что в топик mpg.public.measurements попали новые данные.

Удалите созданные ресурсы

Если созданные ресурсы вам больше не нужны, удалите их:

1. Удалите виртуальную машину.

2. Если вы зарезервировали для виртуальной машины публичный статический IP-адрес, удалите его.

3. Удалите кластеры:

   • Managed Service for PostgreSQL;
   • Managed Service for Apache Kafka®.