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

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

Статья создана

В кластер Managed Service for YDB можно в реальном времени поставлять данные из топиков Apache Kafka®. Эти данные будут автоматически добавлены в таблицы YDB с именами топиков.

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

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

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

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

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

    1. Создайте кластер-источник Managed Service for Apache Kafka® любой подходящей конфигурации.
    2. Создайте базу данных Managed Service for YDB любой подходящей конфигурации.
    3. Создайте в кластере-источнике топик с именем sensors.
    4. Создайте в кластере-источнике пользователя с правами доступа ACCESS_ROLE_PRODUCER, ACCESS_ROLE_CONSUMER к созданному топику.

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

    2. Скачайте файл с настройками провайдера. Поместите его в отдельную рабочую директорию и укажите значения параметров.

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

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

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

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

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

    5. Выполните команду terraform init в директории с конфигурационным файлом. Эта команда инициализирует провайдер, указанный в конфигурационных файлах, и позволяет работать с ресурсами и источниками данных провайдера.

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

      terraform validate

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

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

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

        terraform plan

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

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

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

          terraform apply

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

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

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

    В созданный топик Apache Kafka® sensors кластера-источника будут поступать тестовые данные от сенсоров автомобиля в формате JSON:

    {
    "device_id":"iv9a94th6rztooxh5ur2",
    "datetime":"2020-06-05 17:27:00",
    "latitude":"55.70329032",
    "longitude":"37.65472196",
    "altitude":"427.5",
    "speed":"0",
    "battery_voltage":"23.5",
    "cabin_temperature":"17",
    "fuel_level":null
}

  2. Установите утилиты:

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

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

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

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

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

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

    • Тип базы данных — Apache Kafka®.
    • Параметры эндпоинта:

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

        Выберите кластер-источник из списка и укажите настройки подключения к нему.

      • Расширенные настройки → Правила конвертации.

        • Формат данных – JSON.
        • Схема данных – Вы можете задать схему двумя способами:

          • Список полей.

            Задайте список полей топика вручную:

            Имя Тип Ключ
            device_id STRING Да
            datetime STRING
            latitude DOUBLE
            longitude DOUBLE
            altitude DOUBLE
            speed DOUBLE
            battery_voltage DOUBLE
            cabin_temperature UINT16
            fuel_level UINT16

          • JSON спецификация.

            Создайте и загрузите файл схемы данных в формате JSON json_schema.json:

            json_schema.json
            [
    {
        "name": "device_id",
        "type": "string",
        "key": true
    },
    {
        "name": "datetime",
        "type": "string"
    },
    {
        "name": "latitude",
        "type": "double"
    },
    {
        "name": "longitude",
        "type": "double"
    },
    {
        "name": "altitude",
        "type": "double"
    },
    {
        "name": "speed",
        "type": "double"
    },
    {
        "name": "battery_voltage",
        "type": "double"
    },
    {
        "name": "cabin_temperature",
        "type": "uint16"
    },
    {
        "name": "fuel_level",
        "type": "uint16"
    }
]

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

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

    1. Укажите в файле data-transfer-mkf-ydb.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. Убедитесь, что в базу данных Managed Service for YDB переносятся данные из топика кластера-источника Managed Service for Apache Kafka®:

    1. Создайте файл sample.json с тестовыми данными:

      {
    "device_id": "iv9a94th6rztooxh5ur2",
    "datetime": "2020-06-05 17:27:00",
    "latitude": 55.70329032,
    "longitude": 37.65472196,
    "altitude": 427.5,
    "speed": 0,
    "battery_voltage": 23.5,
    "cabin_temperature": 17,
    "fuel_level": null
}

{
    "device_id": "rhibbh3y08qmz3sdbrbu",
    "datetime": "2020-06-06 09:49:54",
    "latitude": 55.71294467,
    "longitude": 37.66542005,
    "altitude": 429.13,
    "speed": 55.5,
    "battery_voltage": null,
    "cabin_temperature": 18,
    "fuel_level": 32
}

{
    "device_id": "iv9a94th6rztooxh5ur2",
    "datetime": "2020-06-07 15:00:10",
    "latitude": 55.70985913,
    "longitude": 37.62141918,
    "altitude": 417.0,
    "speed": 15.7,
    "battery_voltage": 10.3,
    "cabin_temperature": 17,
    "fuel_level": null
}

    2. Отправьте данные из файла sample.json в топик sensors Managed Service for Apache Kafka® с помощью утилит jq и kafkacat:

      jq -rc . sample.json | kafkacat -P \
   -b <FQDN хоста-брокера>:9091 \
   -t sensors \
   -k key \
   -X security.protocol=SASL_SSL \
   -X sasl.mechanisms=SCRAM-SHA-512 \
   -X sasl.username="<имя пользователя в кластере-источнике>" \
   -X sasl.password="<пароль пользователя в кластере-источнике>" \
   -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexCA.crt -Z

      Данные отправляются от имени созданного пользователя. Подробнее о настройке SSL-сертификата и работе с kafkacat см. в разделе Подключение к топикам в кластере Apache Kafka®.

    3. Убедитесь, что в базу данных Managed Service for YDB перенеслись данные из кластера-источника Managed Service for Apache Kafka®:

      1. В консоли управления выберите каталог, в котором находится нужная база данных.
      2. В списке сервисов выберите Managed Service for YDB.
      3. Выберите базу из списка.
      4. Перейдите на вкладку Навигация.
      5. Проверьте, что база данных Managed Service for YDB содержит таблицу sensors с тестовыми данными из топика.

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

      2. Проверьте, что база данных содержит таблицу sensors с тестовыми данными из топика:

        ydb table query execute \
  --query "SELECT * \
  FROM sensors"

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

Примечание

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

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

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

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

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

  2. Удалите конфигурационный файл data-transfer-mkf-ydb.tf.

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

    terraform validate

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

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

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

      terraform plan

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

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

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

        terraform apply

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

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

    Все ресурсы, которые были описаны в конфигурационном файле data-transfer-mkf-ydb.tf, будут удалены.

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