Захват изменений из PostgreSQL и поставка в Yandex Data Streams

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

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

1. Подготовьте трансфер.
2. Активируйте трансфер.
3. Проверьте работу репликации.

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

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

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

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

1. Создайте поток данных Data Streams с именем mpg-stream.

2. Подключитесь к кластеру Managed Service for PostgreSQL, создайте в базе данных db1 таблицу measurements и заполните ее данными:

   CREATE TABLE measurements (
       device_id varchar(200) 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,
       PRIMARY KEY (device_id)
   );
   INSERT INTO measurements VALUES
       ('iv9a94th6rzt********', '2022-06-05 17:27:00', 55.70329032, 37.65472196,  427.5,    0, 23.5, 17, NULL),
       ('rhibbh3y08qm********', '2022-06-06 09:49:54', 55.71294467, 37.66542005, 429.13, 55.5, NULL, 18, 32);
   

3. Создайте эндпоинт-приемник типа Data Streams со следующими настройками:

   • База данных — ydb-example.
   • Поток — mpg-stream.
   • Сервисный аккаунт — yds-sa.

4. Создайте эндпоинт-источник и трансфер:

   Вручную

   Terraform

   1. Создайте эндпоинт-источник типа PostgreSQL и укажите в нем параметры подключения к кластеру:

      • Тип инсталляции — Кластер Managed Service for PostgreSQL.
      • Кластер Managed Service for PostgreSQL — <имя_кластера-источника_PostgreSQL> из выпадающего списка.
      • База данных — db1.
      • Пользователь — pg-user.
      • Пароль — пароль пользователя pg-user.

   2. Создайте трансфер типа Репликация, использующий созданные эндпоинты.

   1. Укажите в файле postgresql-yds.tf переменные:

      • yds_endpoint_id — значение идентификатора эндпоинта для приемника;
      • transfer_enabled – значение 1 для создания трансфера.

   2. Проверьте корректность файлов конфигурации Terraform с помощью команды:

      terraform validate
      

      Если в файлах конфигурации есть ошибки, Terraform на них укажет.

   3. Создайте необходимую инфраструктуру:

      1. Выполните команду для просмотра планируемых изменений:

         terraform plan
         

         Если конфигурации ресурсов описаны верно, в терминале отобразится список изменяемых ресурсов и их параметров. Это проверочный этап: ресурсы не будут изменены.

      2. Если вас устраивают планируемые изменения, внесите их:

         1. Выполните команду:

            terraform apply
            

         2. Подтвердите изменение ресурсов.

         3. Дождитесь завершения операции.

Активируйте трансферАктивируйте трансфер

1. Активируйте трансфер и дождитесь его перехода в статус Реплицируется.

2. Проверьте, что в поток данных Data Streams перенеслись данные из источника:

   Консоль управления

   AWS CLI

   1. В консоли управления выберите сервис Data Streams.
   2. Выберите поток-приемник из списка и перейдите в раздел  Просмотр данных.
   3. Убедитесь, что в сегменте shard-000000 появились сообщения, содержащие строки таблицы из источника. Чтобы рассмотреть сообщения подробнее, нажмите на значок .

   1. Установите AWS CLI.

   2. Настройте окружение для Data Streams.

   3. Прочитайте данные из потока с помощью:

      • AWS CLI.
      • AWS SDK.

Проверьте работу репликацииПроверьте работу репликации

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

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

   INSERT INTO measurements VALUES
       ('ad02l5ck6sdt********', '2022-06-05 17:27:00', 55.70329032, 37.65472196,  427.5,    0, 23.5, 19, 45);
   

3. Убедитесь, что новая строка отобразилась в потоке данных Data Streams.

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

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

1. Удалите трансфер.
2. Удалите эндпоинт-приемник.
3. Удалите поток данных Data Streams.

Остальные ресурсы удалите в зависимости от способа их создания: