Получение данных из Yandex Managed Service for Apache Kafka® в Yandex Managed Service for ClickHouse®

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

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

1. Настройте интеграцию с Apache Kafka® для кластера Managed Service for ClickHouse®.
2. Создайте в кластере Managed Service for ClickHouse® таблицы на движке Kafka.
3. Отправьте тестовые данные в топики Managed Service for Apache Kafka®.
4. Проверьте наличие тестовых данных в таблицах кластера Managed Service for ClickHouse®.

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

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

В стоимость поддержки описываемого решения входят:

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

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

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

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

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://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
     

Настройте интеграцию с Apache Kafka® для кластера Managed Service for ClickHouse®Настройте интеграцию с Apache Kafka® для кластера Managed Service for ClickHouse®

Создайте в кластере Managed Service for ClickHouse® таблицы на движке KafkaСоздайте в кластере Managed Service for ClickHouse® таблицы на движке Kafka

Пусть в топики Apache Kafka® поступают некоторые данные от сенсоров автомобиля в формате JSON. Эти данные будут передаваться в виде сообщений Apache Kafka®, каждое из которых будет содержать строчку вида:

{"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® будет использовать при вставке в таблицы на движке Kafka формат данных JSONEachRow, который преобразует строки из сообщения Apache Kafka® в нужные значения столбцов.

Для каждого из топиков Apache Kafka® создайте в кластере Managed Service for ClickHouse® отдельную таблицу, куда будут заноситься поступающие данные:

1. Подключитесь к базе данных db1 кластера Managed Service for ClickHouse® с помощью clickhouse-client.

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

   Один кластер Apache Kafka®

   Несколько кластеров Apache Kafka®

   CREATE TABLE IF NOT EXISTS db1.<имя_таблицы_для_топика>
   (
       device_id String,
       datetime DateTime,
       latitude Float32,
       longitude Float32,
       altitude Float32,
       speed Float32,
       battery_voltage Nullable(Float32),
       cabin_temperature Float32,
       fuel_level Nullable(Float32)
   ) ENGINE = Kafka()
   SETTINGS
       kafka_broker_list = '<FQDN_хоста-брокера>:9091',
       kafka_topic_list = '<имя_топика>',
       kafka_group_name = 'sample_group',
       kafka_format = 'JSONEachRow';
   

   CREATE TABLE IF NOT EXISTS db1.<имя_таблицы_для_топика>
   (
       device_id String,
       datetime DateTime,
       latitude Float32,
       longitude Float32,
       altitude Float32,
       speed Float32,
       battery_voltage Nullable(Float32),
       cabin_temperature Float32,
       fuel_level Nullable(Float32)
   ) ENGINE = Kafka(<имя_коллекции_с_данными_аутентификации>);
   

Созданные таблицы будут автоматически наполняться сообщениями, считываемыми из топиков Managed Service for Apache Kafka®. При чтении данных Managed Service for ClickHouse® использует указанные ранее настройки для пользователей с ролью ACCESS_ROLE_CONSUMER.

Подробнее о создании таблиц на движке Kafka см. в документации ClickHouse®.

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

1. Создайте файл 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
   }
   

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/RootCA.crt -Z
   

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

Проверьте наличие тестовых данных в таблицах кластера Managed Service for ClickHouse®Проверьте наличие тестовых данных в таблицах кластера Managed Service for ClickHouse®

Для доступа к данным используйте материализованное представление. Когда к таблице на движке Kafka присоединяется материализованное представление, оно начинает в фоновом режиме собирать данные. Это позволяет непрерывно получать сообщения от Apache Kafka® и преобразовывать их в необходимый формат с помощью SELECT.

Чтобы создать такое представление:

1. Подключитесь к базе данных db1 кластера Managed Service for ClickHouse® с помощью clickhouse-client.

2. Для каждой таблицы на движке Kafka выполните запросы:

   CREATE TABLE db1.temp_<имя_таблицы_для_топика>
   (
       device_id String,
       datetime DateTime,
       latitude Float32,
       longitude Float32,
       altitude Float32,
       speed Float32,
       battery_voltage Nullable(Float32),
       cabin_temperature Float32,
       fuel_level Nullable(Float32)
   ) ENGINE = MergeTree()
   ORDER BY device_id;
   

   CREATE MATERIALIZED VIEW db1.<имя_представления> TO db1.temp_<имя_таблицы_для_топика>
       AS SELECT * FROM db1.<имя_таблицы_для_топика>;
   

Чтобы получить все данные из нужного материализованного представления:

1. Подключитесь к базе данных db1 кластера Managed Service for ClickHouse® с помощью clickhouse-client.

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

   SELECT * FROM db1.<имя_представления>;
   

Запрос вернет таблицу с данными, отправленными в соответствующий топик Managed Service for Apache Kafka®.

Подробнее о работе с данными, поставляемыми из Apache Kafka®, см. в документации ClickHouse®.

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

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

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