集成 EMQX 与 ClickHouse

EMQX 与 ClickHouse 集成简介

​

连接 EMQX

EMQX 是一款开源 MQTT broker，拥有高性能的实时消息处理引擎，可为海量 IoT 设备提供事件流能力。作为扩展能力最强的 MQTT broker，EMQX 可帮助您连接任意设备，轻松应对任意规模，并可在任何地方传输和处理 IoT 数据。

EMQX Cloud 是由 EMQ 提供的、面向 IoT 领域的 MQTT 消息中间件产品。作为全球首个 fully managed 的 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 broker。

​

获取 ClickHouse Cloud 服务

在此设置过程中，我们将 ClickHouse 实例部署在 AWS 的 N. Virginia (us-east-1) ，同时也在同一区域部署了一个 EMQX Cloud 实例。

在设置过程中，你还需要留意连接设置。在本教程中，我们选择 “Anywhere”；但如果你申请了特定位置，则需要将从 EMQX Cloud 部署中获取的 NAT gateway IP 地址添加到白名单。

然后，你需要保存用户名和密码，以备后续使用。

完成后，你将获得一个正在运行的 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)

​

在 EMQX Cloud 上创建 MQTT 服务

在 EMQX Cloud 上创建专用 MQTT broker，只需轻点几下即可。

​

获取账户

EMQX Cloud 为每个账户的标准部署和专业部署均提供 14 天免费试用。

如果您是首次使用 EMQX Cloud，请前往 EMQX Cloud 注册 页面，点击 Start Free 注册账户。

​

创建 MQTT 集群

登录后，点击账户菜单中的“Cloud console”，即可看到用于创建新部署的绿色按钮。

在本教程中，我们将使用 Professional 部署，因为只有 Pro 版本提供数据集成功能，无需编写任何代码即可将 MQTT 数据直接发送到 ClickHouse。

选择 Pro 版本，选择 N.Virginial 区域，然后点击 Create Now。几分钟后，你就会获得一个全托管的 MQTT broker：

现在点击该面板进入集群视图。在此仪表板上，你将看到 MQTT broker 的概览。

​

添加客户端凭据

EMQX Cloud 默认不允许匿名连接，因此你需要添加客户端凭据，才能使用 MQTT 客户端工具向该 broker 发送数据。

点击左侧菜单中的“Authentication & ACL”，然后点击子菜单中的“Authentication”。点击右侧的“Add”按钮，并为稍后建立 MQTT 连接设置用户名和密码。这里我们将使用 emqx 和 xxxxxx 作为用户名和密码。

点击“Confirm”后，一个可用的全托管 MQTT broker 就准备好了。

​

启用 NAT gateway

在开始设置 ClickHouse 集成之前，需要先启用 NAT gateway。默认情况下，MQTT broker 部署在私有 VPC 中，无法通过公网向第三方系统发送数据。

返回 Overview 页面并滚动到页面底部，你会看到 NAT gateway 组件。点击 Subscribe 按钮并按照提示操作。请注意，NAT Gateway 是一项增值服务，但提供 14 天免费试用。

创建完成后，你可以在该组件中看到公网 IP 地址。请注意，如果你在设置 ClickHouse Cloud 时选择了 “Connect from a specific location”，则需要将该 IP 地址添加到白名单中。

​

将 EMQX Cloud 与 ClickHouse Cloud 集成

EMQX Cloud 数据集成 用于配置处理和响应 EMQX 消息流及设备事件的规则。数据集成不仅提供了清晰、灵活且“可配置”的架构方案，还简化了开发流程，提升了易用性，并降低了业务系统与 EMQX Cloud 之间的耦合度。它还为定制 EMQX Cloud 的专有能力提供了更完善的基础设施。

EMQX Cloud 为常见数据系统提供了 30 多种原生集成，ClickHouse 就是其中之一。

​

创建 ClickHouse 资源

点击左侧菜单中的“Data Integrations”，然后点击“View All Resources”。你可以在 Data Persistence 部分找到 ClickHouse，也可以直接搜索 ClickHouse。

点击 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 topic 读取消息，并为该 JSON 对象补充 client_id、topic 和时间戳信息。

因此，你发送到该 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”。现在，一切应该都已设置妥当。您可以在规则详情页面查看数据集成是否正常工作。

所有发送到 temp_hum/emqx topic 的 MQTT 消息都会被持久化到您的 ClickHouse Cloud 数据库中。

​

将数据写入 ClickHouse

我们将模拟温度和湿度数据，通过 MQTT X 将这些数据上报到 EMQX Cloud，然后使用 EMQX Cloud Data Integrations 将数据写入 ClickHouse Cloud。

​

向 EMQX Cloud 发布 MQTT 消息

你可以使用任意 MQTT 客户端或 SDK 发布消息。在本教程中，我们将使用 MQTT X，这是 EMQ 提供的一款易于使用的 MQTT 客户端应用。

在 MQTTX 中点击 “New Connection”，并填写连接表单：

• Name：连接名称。可使用任意你喜欢的名称。
• Host：MQTT broker 的连接地址。可从 EMQX Cloud 的 Overview 页面获取。
• Port：MQTT broker 的连接端口。可从 EMQX Cloud 的 Overview 页面获取。
• Username/Password：使用上面创建的凭证，本教程中应为 emqx 和 xxxxxx。

点击右上角的 “Connect” 按钮后，即可建立连接。

现在，你可以使用此工具向 MQTT broker 发送消息。 输入如下：

1. 将载荷格式设为 “JSON”。
2. 将 topic 设为：temp_hum/emqx (即我们刚刚在规则中设置的 topic)
3. JSON 正文：

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

点击右侧的发送按钮。你可以修改温度值，并向 MQTT broker 发送更多数据。

发送到 EMQX Cloud 的数据会由规则引擎处理，并自动写入 ClickHouse Cloud。

​

查看规则监控

检查规则监控，确认成功次数增加了 1。

​

检查已持久化的数据

现在可以查看 ClickHouse Cloud 中的数据了。理想情况下，你通过 MQTTX 发送的数据会先进入 EMQX Cloud，并借助原生数据集成持久化到 ClickHouse Cloud 的 database 中。

你可以连接到 ClickHouse Cloud 面板中的 SQL 控制台，或使用任意客户端工具从 ClickHouse 中查询数据。本教程中，我们使用的是 SQL 控制台。 执行以下 SQL：

SELECT * FROM emqx.temp_hum;

​

总结

你无需编写任何代码，就已经让 MQTT 数据从 EMQX Cloud 流入 ClickHouse Cloud。借助 EMQX Cloud 和 ClickHouse Cloud，你无需管理基础设施，只需专注于开发 IoT 应用，数据则可安全地存储在 ClickHouse Cloud 中。