Связаться с намиПодключиться

Захват изменений из YDB и поставка в Apache Kafka®

Статья создана
Обновлена 2 декабря 2024 г.

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

Примечание

В YDB CDC-режим поддерживается, начиная с версии 22.5 и выше.

Чтобы запустить поставку данных:

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

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

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

  1. Подготовьте инфраструктуру поставки данных:

    1. Создайте базу данных Managed Service for YDB любой подходящей конфигурации.

    2. Если вы выбрали режим БД Dedicated, создайте и настройте группу безопасности в сети, где находится БД.

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

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

    5. Настройте в кластере-приемнике топики Apache Kafka®. Настройки различаются в зависимости от используемого способа управления топиками. Имена топиков для данных формируются как <префикс_топика>.<имя_таблицы_YDB>. В этом руководстве в качестве примера используется префикс cdc.

      • Если выбрано управление топиками через стандартные интерфейсы Yandex Cloud (Консоль управления, CLI, API):

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

          Для отслеживания изменений в нескольких таблицах создайте для каждой из них отдельный топик с префиксом cdc.

        2. Создайте пользователя с ролями ACCESS_ROLE_CONSUMER и ACCESS_ROLE_PRODUCER для топика cdc.sensors. Чтобы включить все созданные топики, укажите в имени топика cdc.*.

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

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

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

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

    1. Если у вас еще нет Terraform, установите его.

    2. Получите данные для аутентификации. Вы можете добавить их в переменные окружения или указать далее в файле с настройками провайдера.

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

    4. Поместите конфигурационный файл в отдельную рабочую директорию и укажите значения параметров. Если данные для аутентификации не были добавлены в переменные окружения, укажите их в конфигурационном файле.

    5. Скачайте в ту же рабочую директорию файл конфигурации data-transfer-ydb-mkf.tf.

      В этом файле описаны:

      • сеть;
      • подсеть;
      • группа безопасности и правило, необходимое для подключения к кластеру Managed Service for Apache Kafka®;
      • база данных Managed Service for YDB;
      • кластер-приемник Managed Service for Apache Kafka®;
      • топик Apache Kafka®;
      • пользователь Apache Kafka®;
      • трансфер.

      Выбор способа управления топиками определяется переменной Terraform kf_topics_management. Переменная задается при выполнении команд terraform plan и terraform apply (см. далее):

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

        1. Для отслеживания изменений в нескольких таблицах добавьте в файл конфигурации для каждой из них описание отдельного топика с префиксом cdc.
        2. Задайте для переменной Terraform kf_topics_management значение false.

      • Если для управления топиками используется Kafka Admin API, задайте для переменной Terraform kf_topics_management значение true.

    6. Укажите в файле data-transfer-ydb-mkf.tf переменные:

      • source_db_name — имя базы данных Managed Service for YDB;
      • target_kf_version – версия Apache Kafka® в кластере-приемнике;
      • target_user_name – имя пользователя для подключения к топику Apache Kafka®;
      • target_user_password – пароль пользователя;
      • transfer_enabled – значение 0, чтобы не создавать трансфер до создания эндпоинтов вручную.

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

      terraform validate

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

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

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

        terraform plan

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

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

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

          terraform apply

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

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

      В указанном каталоге будут созданы все требуемые ресурсы. Проверить появление ресурсов и их настройки можно в консоли управления.

  2. Установите утилиту kafkacat для чтения и записи данных в топики Apache Kafka®.

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

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

Подготовьте источник

  1. Подключитесь к базе данных Managed Service for YDB.

  2. Создайте YDB-таблицу. В качестве примера используется таблица sensors с информацией, поступающей от условных датчиков автомобиля.

    Добавьте в таблицу колонки вручную:

    Имя Тип Первичный ключ
    device_id String Да
    datetime String
    latitude Double
    longitude Double
    altitude Double
    speed Double
    battery_voltage Double
    cabin_temperature Uint8
    fuel_level Uint32

    Остальные настройки оставьте по умолчанию.

    Также создать таблицу можно YQL-командой:

    CREATE TABLE sensors (
    device_id String,
    datetime String,
    latitude Double,
    longitude Double,
    altitude Double,
    speed Double,
    battery_voltage Double,
    cabin_temperature Uint8,
    fuel_level Uint32,
    PRIMARY KEY (device_id)
)

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

  1. Создайте эндпоинт для источника:

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

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

      • Настройки подключения:

        • База данных — выберите базу данных Managed Service for YDB из списка.
        • Идентификатор сервисного аккаунта — выберите или создайте сервисный аккаунт с ролью editor.

      • Список включенных путей — укажите имена таблиц и директорий базы данных Managed Service for YDB для переноса.

        Важно

        Реплицируются только указанные таблицы и директории. Если не указать имен, то никакие таблицы не будут перенесены.

  2. Создайте эндпоинт для приемника:

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

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

      • Тип подключения — Кластер Managed Service for Apache Kafka.

        • Кластер Managed Service for Apache Kafka — выберите созданный ранее кластер-источник Managed Service for Apache Kafka®.
        • Аутентификация — укажите данные созданного ранее пользователя Apache Kafka®.

      • Топик — Полное имя топика.

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

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

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

  3. Создайте трансфер:

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

    1. Укажите в файле data-transfer-ydb-mkf.tf переменные:

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

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

      terraform validate

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

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

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

        terraform plan

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

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

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

          terraform apply

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

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

      Трансфер активируется автоматически после создания.

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

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

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

    kafkacat \
    -C \
    -b <FQDN_хоста-брокера_1>:9091,...,<FQDN_хоста-брокера_N>:9091 \
    -t cdc.sensors \
    -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®.

  3. Подключитесь к базе данных Managed Service for YDB и добавьте тестовые данные в таблицу sensors:

    REPLACE INTO sensors (device_id, datetime, latitude, longitude, altitude, speed, battery_voltage, cabin_temperature, fuel_level) 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),
    ('iv9a94th6rzt********', '2022-06-08 17:45:00', 53.70987913, 36.62549834, 378.0, NULL, 20.5, 15, 20);

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

    Пример фрагмента сообщения
    {
  "payload": {
      "device_id": "aXY5YTk0dGg2cnp0b294********"
    },
    "schema": {
      "fields": [
        {
            "field": "device_id",
            "optional": false,
            "type": "bytes"
        }
      ],
      "name": "cdc..sensors.Key",
      "optional": false,
      "type": "struct"
    }
}: {
  "payload": {
    "after": {
        "altitude": 378,
        "battery_voltage": 20.5,
        "cabin_temperature": 15,
        "datetime": "MjAyMi0wNi0wOCAxNzo0********",
        "device_id": "aXY5YTk0dGg2cnp0b294********",
        "fuel_level": 20,
        "latitude": 53.70987913,
        "longitude": 36.62549834,
        "speed": null
    },
    "before": null,
    "op": "c",
    "source": {
        "db": "",
        "name": "cdc",
        "snapshot": "false",
        "table": "sensors",
        "ts_ms": 1678642104797,
        "version": "1.1.2.Final"
    },
    "transaction": null,
    "ts_ms": 1678642104797
  },
  "schema": {
    "fields": [
        {
            "field": "before",
            "fields": [
                {
                    "field": "device_id",
                    "optional": false,
                    "type": "bytes"
                },
                ...
            ],
            "name": "cdc..sensors.Value",
            "optional": true,
            "type": "struct"
        },
        {
            "field": "after",
            "fields": [
                {
                    "field": "device_id",
                    "optional": false,
                    "type": "bytes"
                },
                ...
            ],
            "name": "cdc..sensors.Value",
            "optional": true,
            "type": "struct"
        },
        {
            "field": "source",
            "fields": [
                {
                    "field": "version",
                    "optional": false,
                    "type": "string"
                },
                {
                    "field": "connector",
                    "optional": false,
                    "type": "string"
                },
                {
                    "field": "name",
                    "optional": false,
                    "type": "string"
                },
                {
                    "field": "ts_ms",
                    "optional": false,
                    "type": "int64"
                },
                {
                    "default": "false",
                    "field": "snapshot",
                    "name": "io.debezium.data.Enum",
                    "optional": true,
                    "parameters": {
                        "allowed": "true,last,false"
                    },
                    "type": "string",
                    "version": 1
                },
                {
                    "field": "db",
                    "optional": false,
                    "type": "string"
                },
                {
                    "field": "table",
                    "optional": false,
                    "type": "string"
                }
            ],
            "optional": false,
            "type": "struct"
        },
        ...,
        {
            "field": "transaction",
            "fields": [
                {
                    "field": "id",
                    "optional": false,
                    "type": "string"
                },
                {
                    "field": "total_order",
                    "optional": false,
                    "type": "int64"
                },
                {
                    "field": "data_collection_order",
                    "optional": false,
                    "type": "int64"
                }
            ],
            "optional": true,
            "type": "struct"
        }
    ],
    "name": "cdc..sensors.Envelope",
    "optional": false,
    "type": "struct"
  }
}

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

Примечание

Перед тем как удалить созданные ресурсы, деактивируйте трансфер.

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

  1. Удалите трансфер.
  2. Удалите эндпоинты для источника и приемника.
  3. Если при создании эндпоинта для источника вы создавали сервисный аккаунт, удалите его.

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

  1. В терминале перейдите в директорию с планом инфраструктуры.

    Важно

    Убедитесь, что в директории нет Terraform-манифестов с ресурсами, которые вы хотите сохранить. Terraform удаляет все ресурсы, которые были созданы с помощью манифестов в текущей директории.

  2. Удалите ресурсы:

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

      terraform destroy

    2. Подтвердите удаление ресурсов и дождитесь завершения операции.

    Все ресурсы, которые были описаны в Terraform-манифестах, будут удалены.

Предыдущая
PostgreSQL в YDS
Следующая
YDB и поставка в YDS