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

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

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

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

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

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

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

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

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

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

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

1. Подключитесь к виртуальной машине по SSH.

   1. Установите RabbitMQ:

      sudo apt update && sudo apt install rabbitmq-server --yes
      

   2. Создайте пользователя для RabbitMQ:

      sudo rabbitmqctl add_user <имя_пользователя> <пароль>
      

   3. Выдайте этому пользователю права на подключение к серверу:

      sudo rabbitmqctl set_permissions -p / <имя_пользователя> ".*" ".*" ".*" && \
      sudo rabbitmqctl set_topic_permissions -p / <имя_пользователя> amq.topic "cars" "cars"
      

2. Установите утилиты amqp-publish и amqp-declare-queue для работы с RabbitMQ и jq для потоковой обработки JSON-файлов:

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

3. Убедитесь, что можете создать в RabbitMQ очередь cars с помощью amqp-declare-queue:

   amqp-declare-queue \
       --url=amqp://<имя_пользователя>:<пароль>@<IP-адрес_или_FQDN_сервера_RabbitMQ>:5672 \
       --queue=cars
   

4. Установите утилиту 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 clickhouse-client --yes
      

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

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

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

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

resource "yandex_mdb_clickhouse_cluster" "clickhouse-cluster" {
  ...
  clickhouse {
    ...
    config {
      rabbitmq {
        username = "<имя_пользователя>"
        password = "<пароль>"
      }
    }
    ...
  }
}

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

Пусть в RabbitMQ в очередь cars на точке обмена exchange в RabbitMQ будут помещены данные от сенсоров автомобиля в формате JSON:

• строковый идентификатор устройства device_id;

• дата и время формирования данных datetime в формате YYYY-MM-DD HH:MM:SS;

• координаты автомобиля:

  • широта latitude;
  • долгота longitude;
  • высота над уровнем моря altitude;

• текущая скорость speed;

• напряжение батарей battery_voltage (для электромобиля, для автомобиля с ДВС значение этого параметра — null);

• температура в салоне cabin_temperature;

• уровень топлива fuel_level (для автомобиля с ДВС, для электромобиля значение этого параметра — null).

Эти данные будут передаваться в виде сообщений RabbitMQ. Каждое такое сообщение будет содержать 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, который преобразует строки из сообщения RabbitMQ в нужные значения столбцов.

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

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

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

   CREATE TABLE IF NOT EXISTS db1.cars (
       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 = RabbitMQ
   SETTINGS
       rabbitmq_host_port = '<внутренний_IP-адрес_ВМ_с_RabbitMQ>:5672',
       rabbitmq_routing_key_list = 'cars',
       rabbitmq_exchange_name = 'exchange',
       rabbitmq_format = 'JSONEachRow';
   

Эта таблица будет автоматически наполняться сообщениями, считываемыми из очереди cars в точке обмена exchange RabbitMQ. При чтении данных Managed Service for ClickHouse® использует указанные ранее данные для аутентификации.

Отправьте тестовые данные в очередь RabbitMQОтправьте тестовые данные в очередь RabbitMQ

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. Отправьте в созданную ранее очередь cars в точку обмена exchange данные из файла sample.json с помощью jq и amqp-publish.

   jq \
   --raw-output \
   --compact-output . ./sample.json |\
   amqp-publish \
   --url=amqp://<имя_пользователя_RabbitMQ>:<пароль>@<IP-адрес_или_FQDN_сервера_RabbitMQ>:5672 \
   --routing-key=cars \
   --exchange=exchange
   

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

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

Чтобы создать материализованное представление для таблицы db1.cars:

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

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

   CREATE TABLE IF NOT EXISTS db1.cars_data_source (
       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.cars_view TO db1.cars_data_source
     AS SELECT * FROM db1.cars;
   

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

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

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

   SELECT * FROM db1.cars_view;
   

Запрос вернет таблицу с данными, отправленными в RabbitMQ.

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

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

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