Подключиться
Практические руководства

Поставка данных из 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. Проверьте работоспособность трансфера.

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

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

  1. Создайте кластер-источник Managed Service for PostgreSQL любой подходящей конфигурации со следующими настройками:

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

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

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

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

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

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

  1. Чтобы сервис Data Transfer мог получать от кластера Managed Service for PostgreSQL уведомления об изменениях в данных, в кластере-источнике необходимо создать публикацию (publication). Чтобы пользователь pg-user мог создать публикацию, назначьте ему роль mdb_replication.

  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
    ('iv9a94th6rztooxh5ur2', '2020-06-05 17:27:00', 55.70329032, 37.65472196,  427.5,    0, 23.5, 17, NULL),
    ('rhibbh3y08qmz3sdbrbu', '2020-06-06 09:49:54', 55.71294467, 37.66542005, 429.13, 55.5, NULL, 18, 32),
    ('iv9a94th678tooxh5ur2', '2020-06-07 15:00:10', 55.70985913, 37.62141918,  417.0, 15.7, 10.3, 17, NULL);

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

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

Если управление топиками осуществляется с помощью стандартных интерфейсов Yandex Cloud (Консоль управления, YC CLI, Terraform, API):

  1. Создайте топик с именем cdc.public.measurements.

    Если необходимо отслеживать изменения в нескольких таблицах, создайте для каждой из них отдельный топик.

  2. Создайте пользователя с именем kafka-user и ролями ACCESS_ROLE_CONSUMER и ACCESS_ROLE_PRODUCER, действующими на созданные топики.

Если для управления топиками используется Kafka Admin API:

  1. Создайте пользователя-администратора с именем kafka-user.

  2. Помимо роли ACCESS_ROLE_ADMIN назначьте пользователю-администратору роли ACCESS_ROLE_CONSUMER и ACCESS_ROLE_PRODUCER на топики, имена которых начинаются с префикса cdc.

    Необходимые топики будут созданы автоматически при первом событии изменения в отслеживаемых таблицах кластера-источника. Такое решение может быть удобным для отслеживания изменений во множестве таблиц, однако, требует запас свободного места в хранилище кластера. Подробнее см. в разделе Типы дисков в Managed Service for Apache Kafka®.

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

  1. Создайте эндпоинты.

    • Эндпоинт для источника:

      • Тип базы данных — PostgreSQL.
      • Параметры эндпоинта:
        • Настройки подключения — Кластер MDB.
        • Кластер MDB — выберите созданный ранее кластер Managed Service for PostgreSQL.
        • Имя базы данных — db1.
        • Имя пользователя — pg-user.
        • Пароль — укажите пароль пользователя pg-user.
        • Список включенных таблиц — public.measurements.

    • Эндпоинт для приемника:

      • Тип базы данных — Kafka.

      • Параметры эндпоинта:

        • Подключение — Managed Kafka.

        • Managed Kafka:

          • ID кластера Managed Kafka — выберите кластер-приемник.
          • Аутентификация — укажите данные созданного ранее пользователя kafka-user.

        • Настройки кафка-топика — Полное имя топика.

        • Полное имя топика — cdc.public.measurements.

        Если необходимо отслеживать изменения в нескольких таблицах, заполните поля следующим образом:

        • Настройки кафка-топика — Префикс топика.
        • Префикс топика — укажите префикс cdc, использованный при формировании имен топиков.

  2. Создайте трансфер со следующими настройками:

    • Эндпоинты:
      • Источник — созданный ранее эндпоинт для источника.
      • Приемник — созданный ранее эндпоинт для приемника.
    • Тип трансфера — Репликация.

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

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

  1. В отдельном терминале запустите утилиту 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/YandexCA.crt \
    -Z \
    -K:

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

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

    Пример фрагмента сообщения
    {
  "payload": {
    "consumer":"dttuhfpp97l30jaka3ql"
  },
  "schema": {
    "fields": [
      {
        "field": "consumer",
        "optional":false,
        "type":"string"
      }
    ],
    "name": "__data_transfer_stub.public.__consumer_keeper.Key",
    "optional":false,
    "type":"struct"
  }
}:{
  "payload": {
    "after": {
      "consumer":"dttuhfpp97l30jaka3ql",
      "locked_by":"dttuhfpp97l30jaka3ql-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
  },
...

  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. Убедитесь, что в терминале с запущенной утилитой kafkacat отобразились сведения о добавленной строке.

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

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

  1. Деактивируйте и удалите трансфер.

  2. Удалите эндпоинты.

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

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

В этой статье: