Обработка потока изменений Debezium

Debezium — это сервис для захвата изменений в базах данных (Change Data Capture) и отправки их на обработку в другие системы. С помощью Yandex Data Streams можно захватывать эти изменения и отправлять их в Cloud Functions.

Ниже приведена архитектура решения:

НастройкаНастройка

Для получения потока данных необходимо:

1. Создать поток данных Yandex Data Streams.
2. Настроить реквизиты подключения к Yandex Data Streams.
3. Настроить и запустить Debezium Server.
4. Настроить триггер в Cloud Functions для обработки данных.

Создание потока данныхСоздание потока данных

Создайте поток данных Yandex Data Streams с именем debezium. Процедура создания потока данных подробно описана в документации Yandex Data Streams

Настройка реквизитов подключения к Yandex Data StreamsНастройка реквизитов подключения к Yandex Data Streams

1. Войдите в консоль управления. Если вы еще не зарегистрированы, перейдите в консоль управления и следуйте инструкциям.
2. На странице Биллинг убедитесь, что у вас подключен платежный аккаунт, и он находится в статусе ACTIVE или TRIAL_ACTIVE. Если платежного аккаунта нет, создайте его.
3. Если у вас еще нет каталога, создайте его.
4. Создайте сервисный аккаунт и назначьте ему роль editor на ваш каталог.
5. Создайте статический ключ доступа.
6. Настройте AWS CLI:

   1. Установите AWS CLI и выполните команду:

      aws configure
      

   2. Последовательно введите:

      • AWS Access Key ID [None]: — идентификатор ключа сервисного аккаунта.
      • AWS Secret Access Key [None]: — секретный ключ сервисного аккаунта.
      • Default region name [None]: — зону доступности ru-central1.

Настройка Debezium ServerНастройка Debezium Server

В этом примере рассматривается взаимодействие Debezium и PostgreSQL. Далее будем считать, что Debezium будет устанавливаться на сервере, где запущен PostgreSQL.

1. Установите Debezium Server по инструкции.

2. Перейдите в каталог conf и создайте файл application.properties со следующим содержимым:

   debezium.sink.type=kinesis
   debezium.sink.kinesis.region=ru-central1
   debezium.sink.kinesis.endpoint=<эндпоинт>
   debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
   debezium.source.offset.storage.file.filename=data/offsets.dat
   debezium.source.offset.flush.interval.ms=0
   debezium.source.database.hostname=localhost
   debezium.source.database.port=5432
   debezium.source.database.user=<имя_пользователя>
   debezium.source.database.password=<пароль_пользователя>
   debezium.source.database.dbname=<имя_БД>
   debezium.source.database.server.name=debezium
   debezium.source.plugin.name=pgoutput
   
   debezium.source.transforms=Reroute
   debezium.source.transforms.Reroute.type=io.debezium.transforms.ByLogicalTableRouter
   debezium.source.transforms.Reroute.topic.regex=(.*)
   debezium.source.transforms.Reroute.topic.replacement=<поток_данных>
   

   Где:

   • <эндпоинт> — эндпоинт потока данных Data Streams, например, https://yds.serverless.yandexcloud.net/ru-central1/b1g89ae43m6he********/etn01eg4rn1********. Эндпоинт можно посмотреть на странице потока (см. Посмотреть список потоков).
   • <поток_данных> — название потока данных Data Streams.
   • <имя_БД> — название базы данных PostgreSQL.
   • <имя_пользователя> — имя пользователя для подключения к базе данных PostgreSQL.
   • <пароль_пользователя> — пароль пользователя для подключения к базе данных PostgreSQL.

3. Запустите Debezium следующей командой:

   JAVA_OPTS=-Daws.cborEnabled=false ./run.sh
   

4. Выполните какие-либо изменения в базе данных PostgreSQL, например, вставьте данные в таблицу.

5. При правильной настройке в консоли Debezium появятся сообщения вида:

   2022-02-11 07:31:12,850 INFO  [io.deb.con.com.BaseSourceTask] (pool-7-thread-1) 1 records sent during previous 00:19:59.999, last recorded offset: {transaction_id=null, lsn_proc=23576408, lsn_commit=23576120, lsn=23576408, txId=580, ts_usec=1644564672582666}
   

Настроить триггер в Cloud FunctionsНастроить триггер в Cloud Functions

Создайте триггер в Cloud Functions к потоку данных Yandex Data Streams debezium, который был создан выше.

Процедура создания триггера подробно описана в документации Cloud Functions.

В триггер Cloud Functions будут отправлять нотификации обо всех изменениях в базе данных. В коде триггера вы можете обработать эти изменения, реализовав любую необходимую программную обработку.