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

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

Чтобы настроить поставку данных из Managed Service for Apache Kafka® в Managed Service for ClickHouse:

1. Отправьте тестовые данные в топик Managed Service for Apache Kafka®.
2. Подготовьте и активируйте трансфер.
3. Проверьте работоспособность трансфера.

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

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

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

Выполните дополнительные настройки

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

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

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

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

   • clickhouse-client — для подключения к базе данных в кластере Managed Service for ClickHouse.

     1. Подключите DEB-репозиторий ClickHouse:

        sudo apt update && sudo apt install --yes apt-transport-https ca-certificates dirmngr && \
        sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv E0C56BD4 && \
        echo "deb https://repo.clickhouse.com/deb/stable/ main/" | sudo tee \
        /etc/apt/sources.list.d/clickhouse.list
        

     2. Установите зависимости:

        sudo apt update && sudo apt install --yes clickhouse-client
        

     3. Загрузите файл конфигурации для clickhouse-client:

        mkdir --parents ~/.clickhouse-client && \
        wget "https://storage.yandexcloud.net/mdb/clickhouse-client.conf.example" \
            --output-document ~/.clickhouse-client/config.xml
        

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

   • jq — для потоковой обработки JSON-файлов.

     sudo apt update && sudo apt-get install --yes jq
     

Отправьте тестовые данные в топик Managed Service for Apache Kafka®

Пусть в топик Apache Kafka® поступают данные от сенсоров автомобиля. Эти данные будут передаваться в виде сообщений Apache Kafka® в формате 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
}

Кластер Managed Service for ClickHouse будет использовать формат данных JSONEachRow при вставке в таблицы на движке Kafka. Этот формат преобразует строки из сообщения Apache Kafka® в нужные значения столбцов.

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

   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 в топик Managed Service for Apache Kafka® с помощью jq и kafkacat:

   jq -rc . sample.json | kafkacat -P \
      -b <FQDN хоста-брокера>:9091 \
      -t <имя топика> \
      -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
   

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

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

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

   • Параметры эндпоинта → Настройки подключения:

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

       • Кластер Managed Service for Apache Kafka® — выберите кластер-источник из списка.

       • Аутентификация:

         • Имя пользователя — укажите имя пользователя потребителя.
         • Пароль — укажите пароль пользователя потребителя.

     • Полное имя топика — укажите имя топика в кластере Managed Service for Apache Kafka®.

     • (Опционально) Расширенные настройки → Правила конвертации:

       • Формат данных — JSON.

       • Схема данных — JSON спецификация:

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

         json_schema.json

         [
             {
                 "name": "device_id",
                 "type": "string"
             },
             {
                 "name": "datetime",
                 "type": "datetime"
             },
             {
                 "name": "latitude",
                 "type": "double"
             },
             {
                 "name": "longitude",
                 "type": "double"
             },
             {
                 "name": "altitude",
                 "type": "double"
             },
             {
                 "name": "speed",
                 "type": "double"
             },
             {
                 "name": "battery_voltage",
                 "type": "any"
             },
             {
                 "name": "cabin_temperature",
                 "type": "double"
             },
             {
                 "name": "fuel_level",
                 "type": "any"
             }
         ]
         

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

   Вручную

   С помощью Terraform

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

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

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

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

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

            • Кластер Managed Service for ClickHouse — выберите кластер-приемник из списка.

          • База данных — укажите имя базы данных.

          • Пользователь и Пароль — укажите имя и пароль пользователя с доступом к базе, например, владельца базы данных.

        • Расширенные настройки → Загружать данные в формате JSON — включите эту опцию, если в расширенных настройках эндпоинта источника включили опцию Правила конвертации.

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

   3. Активируйте его.

   1. Раскомментируйте в файле data-transfer-mkf-mch.tf:

      • переменную source_endpoint_id и задайте ей значение идентификатора эндпоинта для источника, созданного на предыдущем шаге;
      • ресурсы yandex_datatransfer_endpoint и yandex_datatransfer_transfer.

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

      terraform validate
      

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

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

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

         terraform plan
         

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

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

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

            terraform apply
            

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

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

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

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

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

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

   1. Подключитесь к кластеру с помощью clickhouse-client.

   2. Выполните запрос:

      SELECT * FROM <имя базы данных ClickHouse>.<имя топика Apache Kafka®>
      

3. Измените значения в файле sample.json и отправьте данные из него в топик Managed Service for Apache Kafka®:

   jq -rc . sample.json | kafkacat -P \
      -b <FQDN хоста-брокера>:9091 \
      -t <имя топика> \
      -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
   

4. Убедитесь, что в базе данных Managed Service for ClickHouse отобразились новые значения:

   1. Подключитесь к кластеру с помощью clickhouse-client.

   2. Выполните запрос:

      SELECT * FROM <имя базы данных ClickHouse>.<имя топика Apache Kafka®>
      

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

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

1. Удалите трансфер.
2. Удалите эндпоинт для источника.

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