Поставка данных из очереди Apache Kafka® в ClickHouse®

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

Улучшена

Обновлена 15 декабря 2025 г.

Перед началом работы
Отправьте тестовые данные в топик Managed Service for Apache Kafka®
Подготовьте и активируйте трансфер
Проверьте работоспособность трансфера
Удалите созданные ресурсы

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

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

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

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

Необходимые платные ресурсы

Кластер Managed Service for Apache Kafka®: выделенные хостам вычислительные ресурсы, объем хранилища и резервных копий (см. тарифы Managed Service for Apache Kafka®).
Кластер Managed Service for ClickHouse®: выделенные хостам вычислительных ресурсов, объем хранилища и резервных копий (см. тарифы Managed Service for ClickHouse®).
Публичные IP-адреса, если для хостов кластеров включен публичный доступ (см. тарифы Virtual Private Cloud).

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

Вручную

Terraform

Создайте кластер-источник Managed Service for Apache Kafka® любой подходящей конфигурации. Для подключения к кластеру с локальной машины пользователя, а не из облачной сети Yandex Cloud, включите публичный доступ к кластеру при его создании.
Создайте топик в кластере Managed Service for Apache Kafka®.
Чтобы производитель и потребитель могли работать с топиком в кластере Managed Service for Apache Kafka®, создайте пользователей:
- с ролью ACCESS_ROLE_PRODUCER для производителя;
- с ролью ACCESS_ROLE_CONSUMER для потребителя.
Создайте кластер-приемник Managed Service for ClickHouse® любой подходящей конфигурации. Для подключения к кластеру с локальной машины пользователя, а не из облачной сети Yandex Cloud, включите публичный доступ к кластеру при его создании.
Если вы используете группы безопасности, настройте их так, чтобы к кластерам можно было подключаться из интернета:
- Managed Service for Apache Kafka®.
- Managed Service for ClickHouse®.

Если у вас еще нет Terraform, установите его.
Получите данные для аутентификации. Вы можете добавить их в переменные окружения или указать далее в файле с настройками провайдера.
Настройте и инициализируйте провайдер. Чтобы не создавать конфигурационный файл с настройками провайдера вручную, скачайте его.
Поместите конфигурационный файл в отдельную рабочую директорию и укажите значения параметров. Если данные для аутентификации не были добавлены в переменные окружения, укажите их в конфигурационном файле.
Скачайте в ту же рабочую директорию файл конфигурации data-transfer-mkf-mch.tf.

В этом файле описаны:
- сеть;
- подсеть;
- группа безопасности и правила, необходимые для подключения к кластерам из интернета;
- кластер-источник Managed Service for Apache Kafka®;
- топик и два пользователя Apache Kafka®, от имени которых к топику будут подключаться производитель и потребитель;
- кластер-приемник Managed Service for ClickHouse®;
- эндпоинт для приемника;
- трансфер.
Укажите в файле data-transfer-mkf-mch.tf:
- параметры кластера-источника Managed Service for Apache Kafka®:
  - source_user_producer и source_password_producer — имя и пароль пользователя производителя;
  - source_user_consumer и source_password_consumer — имя и пароль пользователя потребителя;
  - source_topic_name — имя топика;
- параметры кластера-приемника Managed Service for ClickHouse®, которые будут использоваться как параметры эндпоинта-приемника:
  - target_db_name — имя базы данных Managed Service for ClickHouse®;
  - target_user и target_password — имя и пароль пользователя-владельца базы данных.
Проверьте корректность файлов конфигурации Terraform с помощью команды:
```
terraform validate
```
Если в файлах конфигурации есть ошибки, Terraform на них укажет.
Создайте необходимую инфраструктуру:
1. Выполните команду для просмотра планируемых изменений:
```
terraform plan
```
  Если конфигурации ресурсов описаны верно, в терминале отобразится список изменяемых ресурсов и их параметров. Это проверочный этап: ресурсы не будут изменены.
2. Если вас устраивают планируемые изменения, внесите их:
  1. Выполните команду:
```
terraform apply
```
  2. Подтвердите изменение ресурсов.
  3. Дождитесь завершения операции.
В указанном каталоге будут созданы все требуемые ресурсы. Проверить появление ресурсов и их настройки можно в консоли управления.

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

Установите утилиты:
- 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://packages.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 -p ~/.clickhouse-client && \
wget "https://storage.yandexcloud.net/doc-files/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":"iv9a94th6rzt********",
    "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® в нужные значения столбцов.

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

sample.json

{
    "device_id": "iv9a94th6rzt********",
    "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": "rhibbh3y08qm********",
    "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": "iv9a94th6rzt********",
    "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
}

Отправьте данные из файла 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/RootCA.crt -Z

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

Примечание

Для ускорения поставки большого объема данных используйте специальные настройки эндпоинтов.

Создайте эндпоинт для источника:
- Тип базы данных — Kafka.
- Параметры эндпоинта → Настройки подключения:
  - Тип подключения — Кластер Managed Service for Apache Kafka.
    - Кластер Managed Service for Apache Kafka — выберите кластер-источник из списка.
    - Аутентификация:
      - Имя пользователя — укажите имя пользователя потребителя.
      - Пароль — укажите пароль пользователя потребителя.
  - Полное имя топика — укажите имя топика в кластере Managed Service for Apache Kafka®.
  - Расширенные настройки → Правила конвертации:
    - Формат данных — JSON.
    - Схема данных — JSON-спецификация:
      
      Скопируйте и вставьте схему данных в формате 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" } ]
Создайте эндпоинт для приемника и трансфер:
Вручную

Terraform
1. Создайте эндпоинт для приемника:
  - Тип базы данных — ClickHouse.
  - Параметры эндпоинта:
    
    Настройки подключения:
    
    Тип подключения — Managed кластер.
    
    Managed кластер — выберите кластер-приемник из списка.
    
    База данных — укажите имя базы данных.
    
    Пользователь и Пароль — укажите имя и пароль пользователя с доступом к базе, например, владельца базы данных.
    
    Расширенные настройки → Загружать данные в формате 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. Если вас устраивают планируемые изменения, внесите их:
    
    Выполните команду:
    
    terraform apply
    
    Подтвердите изменение ресурсов.
    
    Дождитесь завершения операции.
  Трансфер активируется автоматически после создания.

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

Дождитесь перехода трансфера в статус Реплицируется.
Убедитесь, что в базу данных Managed Service for ClickHouse® перенеслись данные из кластера-источника Managed Service for Apache Kafka®:
1. Подключитесь к кластеру с помощью clickhouse-client.
2. Выполните запрос:
```
SELECT * FROM <имя_базы_данных_ClickHouse®>.<имя_топика_Apache_Kafka>
```

Измените значения в файле 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/RootCA.crt -Z

Убедитесь, что в базе данных Managed Service for ClickHouse® отобразились новые значения:
1. Подключитесь к кластеру с помощью clickhouse-client.
2. Выполните запрос:
```
SELECT * FROM <имя_базы_данных_ClickHouse®>.<имя_топика_Apache_Kafka>
```

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

Примечание

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

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

Удалите трансфер.
Удалите эндпоинт для источника.
Остальные ресурсы удалите в зависимости от способа их создания:
Вручную

Terraform
1. Удалите эндпоинт для приемника.
2. Удалите кластер Managed Service for Apache Kafka®.
3. Удалите кластер Managed Service for ClickHouse®.
1. В терминале перейдите в директорию с планом инфраструктуры.
  
  Важно
  
  Убедитесь, что в директории нет Terraform-манифестов с ресурсами, которые вы хотите сохранить. Terraform удаляет все ресурсы, которые были созданы с помощью манифестов в текущей директории.
2. Удалите ресурсы:
  1. Выполните команду:
    
    terraform destroy
  2. Подтвердите удаление ресурсов и дождитесь завершения операции.
  Все ресурсы, которые были описаны в Terraform-манифестах, будут удалены.

ClickHouse® является зарегистрированным товарным знаком ClickHouse, Inc.

Поставка данных из очереди Apache Kafka® в ClickHouse®

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

Необходимые платные ресурсыНеобходимые платные ресурсы

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

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

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

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

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

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

Была ли статья полезна?