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

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

Чтобы настроить CDC с использованием сервиса Data Transfer:

1. Подготовьте инфраструктуру.
2. Подготовьте кластер-источник.
3. Подготовьте кластер-приемник.
4. Подготовьте и активируйте трансфер.
5. Проверьте работоспособность трансфера.

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

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

Зарегистрируйтесь в Yandex Cloud и создайте платежный аккаунт:

1. Перейдите в консоль управления, затем войдите в Yandex Cloud или зарегистрируйтесь.
2. На странице Yandex Cloud Billing убедитесь, что у вас подключен платежный аккаунт, и он находится в статусе ACTIVE или TRIAL_ACTIVE. Если платежного аккаунта нет, создайте его и привяжите к нему облако.

Если у вас есть активный платежный аккаунт, вы можете создать или выбрать каталог, в котором будет работать ваша инфраструктура, на странице облака.

Подробнее об облаках и каталогах.

Необходимые платные ресурсыНеобходимые платные ресурсы

• Кластер Managed Service for PostgreSQL: выделенные хостам вычислительные ресурсы, объем хранилища и резервных копий (тарифы Managed Service for PostgreSQL).
• Кластер Managed Service for Apache Kafka®: выделенные хостам вычислительные ресурсы, объем хранилища и резервных копий (тарифы Managed Service for Apache Kafka®).
• Публичные IP-адреса, если для хостов кластеров включен публичный доступ (тарифы Yandex Virtual Private Cloud).
• Каждый трансфер: использование вычислительных ресурсов и количество переданных строк данных (тарифы Data Transfer).

Подготовьте инфраструктуруПодготовьте инфраструктуру

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

1. Если вы создавали инфраструктуру вручную, назначьте роль mdb_replication пользователю pg-user. Это необходимо, чтобы пользователь мог создать публикацию, через которую Data Transfer получает информацию об изменениях в данных в кластере-источнике.

2. Подключитесь к базе данных db1 от имени пользователя pg-user.

3. Наполните базу тестовыми данными. В качестве примера используется простая таблица, содержащая информацию с некоторых датчиков автомобиля.

   Создайте таблицу:

   CREATE TABLE public.measurements (
       "device_id" text PRIMARY KEY NOT NULL,
       "datetime" timestamp NOT NULL,
       "latitude" real NOT NULL,
       "longitude" real NOT NULL,
       "altitude" real NOT NULL,
       "speed" real NOT NULL,
       "battery_voltage" real,
       "cabin_temperature" real NOT NULL,
       "fuel_level" real
   );
   

   Наполните таблицу данными:

   INSERT INTO public.measurements VALUES
       ('iv9a94th6rzt********', '2020-06-05 17:27:00', 55.70329032, 37.65472196,  427.5,    0, 23.5, 17, NULL),
       ('rhibbh3y08qm********', '2020-06-06 09:49:54', 55.71294467, 37.66542005, 429.13, 55.5, NULL, 18, 32),
       ('iv9a94th678t********', '2020-06-07 15:00:10', 55.70985913, 37.62141918,  417.0, 15.7, 10.3, 17, NULL);
   

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

Если вы создавали инфраструктуру с помощью Terraform, пропустите этот шаг и перейдите к подготовке и активации трансфера.

Настройка кластера-приемника различается в зависимости от используемого способа управления топиками. При этом имена топиков для данных конструируются по тому же принципу, что и в Debezium — <префикс_топика>.<имя_схемы>.<имя_таблицы>. В этом руководстве в качестве примера будет использоваться префикс cdc.

Подготовьте и активируйте трансферПодготовьте и активируйте трансфер

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

1. Установите на локальный компьютер утилиту kcat (kafkacat) и клиент командной строки PostgreSQL. Например, в Ubuntu 20.04 выполните команду:

   sudo apt update && sudo apt install kafkacat postgresql-client --yes
   

   Убедитесь, что можете с ее помощью подключиться к кластеру-источнику Managed Service for Apache Kafka® через SSL.

2. В отдельном терминале запустите утилиту kafkacat в режиме потребителя:

   kafkacat \
       -C \
       -b <FQDN_хоста-брокера_1>:9091,...,<FQDN_хоста-брокера_N>:9091 \
       -t cdc.public.measurements \
       -X security.protocol=SASL_SSL \
       -X sasl.mechanisms=SCRAM-SHA-512 \
       -X sasl.username=kafka-user \
       -X sasl.password=<пароль> \
       -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
       -Z \
       -K:
   

   FQDN хостов-брокеров можно получить со списком хостов в кластере Managed Service for Apache Kafka®.

   Будет выведена схема формата данных таблицы public.measurements и данные о добавленных в нее ранее строках.

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

   {
     "payload": {
       "consumer":"dttuhfpp97l3********"
     },
     "schema": {
       "fields": [
         {
           "field": "consumer",
           "optional":false,
           "type":"string"
         }
       ],
       "name": "__data_transfer_stub.public.__consumer_keeper.Key",
       "optional":false,
       "type":"struct"
     }
   }:{
     "payload": {
       "after": {
         "consumer":"dttuhfpp97l3********l",
         "locked_by":"dttuhfpp97l3********-1",
         "locked_till":"2022-05-15T09:55:18Z"
       },
     "before": null,
     "op":"u",
     "source": {
       "connector":"postgresql",
       "db":"db1",
       "lsn":85865797008,
       "name":"__data_transfer_stub",
       "schema":"public",
       "snapshot":"false",
       "table":"__consumer_keeper",
       "ts_ms":1652608518883,
       "txId":245165,
       "version":"1.1.2.Final",
       "xmin":null
     },
   ...
   

3. Подключитесь к кластеру-источнику и добавьте данные в таблицу measurements:

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

4. Убедитесь, что в терминале с запущенной утилитой kafkacat отобразились сведения о добавленной строке.

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

Некоторые ресурсы платные. Чтобы за них не списывалась плата, удалите ресурсы, которые вы больше не будете использовать:

Дополнительные материалыДополнительные материалы

Больше информации о сценариях поставок данных в вебинаре Yandex Cloud:

Смотреть видео в Yandex Cloud Video.