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

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 — чрезвычайно легковесным транспортным протоколом обмена сообщениями, построенным по модели publish/subscribe.
• Вы используете EMQX или EMQX Cloud как движок обработки сообщений в реальном времени, обеспечивающий стриминг событий для IoT-устройств в большом масштабе.
• Вы подготовили экземпляр ClickHouse Cloud для хранения данных устройств.
• Мы используем MQTT X в качестве инструмента для тестирования MQTT-клиента, чтобы подключиться к развертыванию EMQX Cloud и публиковать данные MQTT. Также подойдут и другие способы подключения к MQTT-брокеру.

В ходе этой настройки мы развернули экземпляр ClickHouse в AWS в регионе Северная Виргиния (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 можно всего в несколько кликов.

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

Если вы впервые пользуетесь EMQX Cloud, перейдите на страницу регистрации в EMQX Cloud и нажмите Start Free, чтобы создать аккаунт.

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

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

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

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

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

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

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

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

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

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

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

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

Нажмите "Data Integrations" в левом меню, затем — "View All Resources". ClickHouse находится в разделе Data Persistence, либо его можно найти через поиск.

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

• Примечание: добавьте примечание для этого ресурса.
• Адрес сервера: это адрес вашего сервиса ClickHouse Cloud, не забудьте указать порт.
• Имя базы данных: emqx, которую мы создали на предыдущих шагах.
• Пользователь: имя пользователя для подключения к вашему сервису ClickHouse Cloud.
• Ключ: пароль для подключения.

При создании ресурса появится всплывающее окно, и если нажать '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 и временную метку.

Итак, исходный 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.

Мы смоделируем данные о температуре и влажности, отправим их в EMQX Cloud через MQTT X, а затем с помощью Data Integrations в EMQX Cloud сохраним их в ClickHouse 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.