Получение данных из RabbitMQ

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

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

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

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

1. Создайте кластер Managed Service for ClickHouse любой подходящей вам конфигурации с базой данных db1 и публичным доступом ко всем его хостам.

   Примечание

   Интеграцию с RabbitMQ можно настроить уже на этапе создания кластера. В этом сценарии использования интеграция будет настроена позже.

2. Установите утилиты amqp-publish и amqp-declare-queue для работы с RabbitMQ. Убедитесь, что можете создать в RabbitMQ очередь cars с помощью amqp-declare-queue:

   amqp-declare-queue \
       --url=amqp://<имя пользователя RabbitMQ>:<пароль>@<FQDN хоста RabbitMQ>:5672 \
       --queue=cars
   

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

4. Установите утилиту для потоковой обработки JSON-файлов jq.

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

Укажите в настройках кластера Managed Service for ClickHouse имя пользователя и пароль для аутентификации в RabbitMQ (секция Настройки СУБД → Rabbitmq).

Изучите формат данных, поступающих от RabbitMQ

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

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

Создайте в кластере Managed Service for ClickHouse таблицу на движке 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-адрес или FQDN сервера RabbitMQ>:5672',
     rabbitmq_routing_key_list = 'cars',
     rabbitmq_exchange_name = 'exchange',
     rabbitmq_format = 'JSONEachRow';
   

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

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

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

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

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

Проверьте наличие тестовых данных в таблице кластера 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 данные в табличном виде.

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