Интеграция EMQX с ClickHouse

Подключение EMQX

EMQX — это брокер MQTT с открытым исходным кодом и высокопроизводительным движком обработки сообщений в режиме реального времени, обеспечивающий потоковую обработку событий для IoT‑устройств в очень крупном масштабе. Будучи наиболее масштабируемым брокером MQTT, EMQX помогает подключать любые устройства при любом масштабе. Перемещайте и обрабатывайте ваши IoT‑данные где угодно.

EMQX Cloud — это продукт промежуточного слоя обмена сообщениями MQTT для IoT‑сферы, размещаемый компанией EMQ. Как первый в мире полностью управляемый облачный сервис обмена сообщениями MQTT 5.0, EMQX Cloud предоставляет единое решение для эксплуатации и сопровождения, а также уникальную изолированную среду для сервисов обмена сообщениями MQTT. В эпоху Интернета вещей EMQX Cloud помогает быстро создавать отраслевые приложения для IoT‑сферы и легко собирать, передавать, обрабатывать и хранить IoT‑данные.

Используя инфраструктуру облачных провайдеров, EMQX Cloud обслуживает десятки стран и регионов по всему миру, предоставляя недорогие, безопасные и надежные облачные сервисы для приложений 5G и Интернета вещей.

Предварительные условия

• Вы знакомы с протоколом MQTT, который разработан как чрезвычайно легковесный транспортный протокол обмена сообщениями с моделью публикация/подписка.
• Вы используете EMQX или EMQX Cloud в качестве высокопроизводительного движка обработки сообщений в реальном времени, обеспечивающего потоковую обработку событий для IoT‑устройств в крупном масштабе.
• Вы подготовили экземпляр ClickHouse Cloud для долговременного хранения данных устройств.
• Мы используем MQTT X как тестовый MQTT‑клиент для подключения к развертыванию EMQX Cloud и публикации MQTT‑данных. Также подойдут и другие способы подключения к брокеру MQTT.

Получение сервиса ClickHouse Cloud

В ходе этой настройки мы развернули экземпляр ClickHouse в AWS в регионе N. Virginia (us-east -1), при этом экземпляр EMQX Cloud также был развернут в том же регионе.

Во время настройки вам также необходимо обратить внимание на параметры подключения. В этом руководстве мы выбираем "Anywhere", но если вы настраиваете доступ из определенного расположения, вам нужно будет добавить IP-адрес NAT-шлюза, полученный при развертывании EMQX Cloud, в список разрешенных.

Затем сохраните имя пользователя и пароль для дальнейшего использования.

После этого вы получите запущенный экземпляр ClickHouse. Нажмите "Connect", чтобы получить адрес подключения экземпляра ClickHouse Cloud.

Нажмите "Connect to SQL Console", чтобы создать базу данных и таблицу для интеграции с EMQX Cloud.

Вы можете использовать приведенный ниже SQL-запрос или изменить SQL в соответствии с вашими потребностями.

CREATE TABLE emqx.temp_hum
(
   client_id String,
   timestamp DateTime,
   topic String,
   temp Float32,
   hum Float32
)
ENGINE = MergeTree()
PRIMARY KEY (client_id, timestamp)

Создание MQTT-сервиса в EMQX Cloud

Создать выделенный MQTT-брокер в EMQX Cloud так же просто, всего в несколько кликов.

Регистрация учётной записи

EMQX Cloud предоставляет 14-дневную бесплатную пробную версию как для стандартного, так и для профессионального развертывания для каждой учётной записи.

Перейдите на страницу EMQX Cloud sign up и нажмите «Start free», чтобы зарегистрировать учётную запись, если вы впервые используете EMQX Cloud.

Создание MQTT-кластера

После входа в систему нажмите «Cloud console» в меню учётной записи, и вы увидите зелёную кнопку для создания нового развертывания.

В этом руководстве мы будем использовать профессиональное развертывание, потому что только Pro-версия предоставляет функциональность интеграции данных, которая позволяет отправлять MQTT-данные напрямую в ClickHouse без единой строки кода.

Выберите Pro-версию, регион N.Virginial и нажмите Create Now. Всего через несколько минут вы получите полностью управляемый MQTT-брокер:

Теперь нажмите на панель, чтобы перейти на страницу кластера. На этой панели мониторинга вы увидите обзор вашего MQTT-брокера.

Добавление учётных данных клиента

EMQX Cloud по умолчанию не допускает анонимные подключения, поэтому вам нужно добавить учётные данные клиента, чтобы вы могли использовать MQTT-клиент для отправки данных этому брокеру.

Нажмите «Authentication & ACL» в левом меню и выберите «Authentication» в подменю. Нажмите кнопку «Add» справа и задайте имя пользователя и пароль для последующего MQTT-подключения. Здесь мы будем использовать emqx и xxxxxx в качестве имени пользователя и пароля.

Нажмите «Confirm», и теперь у нас есть полностью управляемый MQTT-брокер, готовый к работе.

Включение NAT-шлюза

Прежде чем мы сможем начать настройку интеграции с ClickHouse, нам сначала нужно включить NAT-шлюз. По умолчанию MQTT-брокер развёрнут в приватном VPC, который не может отправлять данные в сторонние системы через общедоступную сеть.

Вернитесь на страницу «Overview» и прокрутите вниз до конца, где вы увидите виджет NAT Gateway. Нажмите кнопку «Subscribe» и следуйте инструкциям. Обратите внимание, что NAT Gateway — это дополнительный платный сервис, но он также предлагает 14-дневный бесплатный пробный период.

После его создания вы найдёте публичный IP-адрес в этом виджете. Обратите внимание, что если вы выбрали «Connect from a specific location» при настройке ClickHouse Cloud, вам нужно будет добавить этот IP-адрес в список разрешённых IP-адресов.

Интеграция EMQX Cloud с ClickHouse Cloud

EMQX Cloud Data Integrations используется для настройки правил обработки и реакции на потоки сообщений EMQX и события устройств. Data Integrations не только предоставляет понятное и гибкое «конфигурируемое» архитектурное решение, но также упрощает процесс разработки, повышает удобство использования и снижает степень связности между бизнес-системой и EMQX Cloud. Кроме того, он обеспечивает превосходную инфраструктуру для кастомизации проприетарных возможностей EMQX Cloud.

EMQX Cloud предлагает более 30 нативных интеграций с популярными системами данных. ClickHouse — одна из них.

Создание ресурса ClickHouse

Нажмите «Data Integrations» в левом меню и затем «View All Resources». Вы найдете ClickHouse в разделе Data Persistence или можете выполнить поиск по ClickHouse.

Нажмите на карточку ClickHouse, чтобы создать новый ресурс.

• Note: добавьте примечание для этого ресурса.
• Server address: это адрес вашего сервиса ClickHouse Cloud, не забудьте указать порт.
• Database name: emqx, базу данных мы создали на предыдущих шагах.
• User: имя пользователя для подключения к вашему сервису ClickHouse Cloud.
• Key: пароль для подключения.

Создание нового правила

Во время создания ресурса вы увидите всплывающее окно, и нажатие кнопки «New» перенаправит вас на страницу создания правила.

EMQX предоставляет мощный движок правил, который может трансформировать и обогащать исходные MQTT-сообщения перед их отправкой во внешние системы.

Ниже приведено правило, используемое в этом руководстве:

SELECT
   clientid AS client_id,
   (timestamp div 1000) AS timestamp,
   topic AS topic,
   payload.temp AS temp,
   payload.hum AS hum
FROM
"temp_hum/emqx"

Он будет читать сообщения из топика temp_hum/emqx и дополнять JSON-объект, добавляя в него client_id, topic и timestamp.

Итак, исходный JSON, который вы отправляете в этот топик:

{"temp": 28.5, "hum": 0.68}

Вы можете использовать SQL‑тест, чтобы проверить запрос и увидеть результаты.

Теперь нажмите кнопку «NEXT». На этом шаге вы указываете EMQX Cloud, как записывать обработанные данные в вашу базу данных ClickHouse.

Добавьте ответное действие

Если у вас только один ресурс, вам не нужно изменять «Resource» и «Action Type». Вам нужно только задать SQL‑шаблон. Ниже приведён пример, используемый в этом руководстве:

INSERT INTO temp_hum (client_id, timestamp, topic, temp, hum) VALUES ('${client_id}', ${timestamp}, '${topic}', ${temp}, ${hum})

Это шаблон для вставки данных в ClickHouse; здесь вы можете увидеть, как используются переменные.

Просмотр сведений о правиле

Нажмите «Confirm» и «View Details». Теперь всё должно быть корректно настроено. На странице сведений о правиле вы можете убедиться, что интеграция данных работает.

Все MQTT-сообщения, отправленные в топик temp_hum/emqx, будут сохранены в вашей базе данных ClickHouse Cloud.

Сохранение данных в ClickHouse

Мы будем имитировать данные о температуре и влажности и отправлять их в EMQX Cloud через MQTT X, а затем использовать EMQX Cloud Data Integrations для сохранения данных в ClickHouse Cloud.

Публикация MQTT‑сообщений в EMQX Cloud

Вы можете использовать любой MQTT‑клиент или SDK для публикации сообщений. В этом руководстве мы будем использовать MQTT X — удобный MQTT‑клиент, предоставляемый EMQ.

Нажмите «New Connection» в MQTTX и заполните форму подключения:

• Name: имя подключения. Можно указать любое удобное имя.
• Host: адрес подключения к MQTT‑брокеру. Его можно получить на странице обзора EMQX Cloud.
• Port: порт подключения MQTT‑брокера. Его можно получить на странице обзора EMQX Cloud.
• Username/Password: используйте созданные выше учетные данные, которые в этом руководстве должны быть emqx и xxxxxx.

Нажмите кнопку «Connect» в правом верхнем углу, после чего подключение должно установиться.

Теперь вы можете отправлять сообщения в MQTT‑брокер, используя этот инструмент. Параметры ввода:

1. Установите формат полезной нагрузки в «JSON».
2. Установите топик: temp_hum/emqx (топик, который мы только что задали в правиле).
3. JSON‑тело:

{"temp": 23.1, "hum": 0.68}

Нажмите кнопку отправки справа. Вы можете изменить значение температуры и отправить больше данных на MQTT‑брокер.

Данные, отправленные в EMQX Cloud, должны быть обработаны движком правил и автоматически вставлены в ClickHouse Cloud.

Просмотр мониторинга правил

Проверьте мониторинг правил и убедитесь, что счётчик успешных срабатываний увеличился на единицу.

Проверка сохранённых данных

Теперь пришло время посмотреть на данные в ClickHouse Cloud. В идеале данные, которые вы отправляете с помощью MQTTX, попадут в EMQX Cloud и будут сохранены в базе данных ClickHouse Cloud с помощью встроенной интеграции данных.

Вы можете подключиться к SQL-консоли в панели ClickHouse Cloud или использовать любой клиентский инструмент, чтобы извлечь данные из вашего ClickHouse. В этом руководстве мы использовали SQL-консоль. Выполнив SQL-запрос:

SELECT * FROM emqx.temp_hum;

Итоги

Вы не написали ни единой строки кода, а данные MQTT уже поступают из EMQX Cloud в ClickHouse Cloud. С EMQX Cloud и ClickHouse Cloud вам не нужно управлять инфраструктурой — вы можете сосредоточиться на разработке IoT‑приложений, пока данные надёжно хранятся в ClickHouse Cloud.