跳至主要内容

将 EMQX 集成到 ClickHouse

连接 EMQX​

EMQX 是一个开源 MQTT 代理，具有高性能的实时消息处理引擎，为大规模物联网设备提供事件流。作为最可扩展的 MQTT 代理，EMQX 可以帮助您连接任何设备，无论规模大小。随时随地移动和处理您的物联网数据。

EMQX Cloud 是由 EMQ 托管的物联网领域的 MQTT 消息中间件产品。作为全球首个完全托管的 MQTT 5.0 云消息服务，EMQX Cloud 提供一站式运维托管和独特的 MQTT 消息服务的隔离环境。在万物互联时代，EMQX Cloud 可以帮助您快速构建物联网领域的行业应用，轻松收集、传输、计算和持久化物联网数据。

借助云提供商提供的基础设施，EMQX Cloud 服务于全球数十个国家和地区，为 5G 和万物互联应用提供低成本、安全、可靠的云服务。

EMQX Cloud Architecture

假设​

  • 您熟悉 MQTT 协议，该协议被设计为一种极其轻量级的发布/订阅消息传输协议。
  • 您正在使用 EMQX 或 EMQX Cloud 作为实时消息处理引擎，为大规模物联网设备提供事件流。
  • 您已准备了一个 Clickhouse Cloud 实例来持久化设备数据。
  • 我们使用 MQTT X 作为 MQTT 客户端测试工具，连接 EMQX Cloud 的部署以发布 MQTT 数据。或者其他连接到 MQTT 代理的方法也可以完成这项工作。

获取您的 ClickHouse Cloud 服务​

在此设置过程中，我们在 AWS 的北弗吉尼亚州 (us-east -1) 部署了 ClickHouse 实例，同时也在同一区域部署了一个 EMQX Cloud 实例。

clickhouse_cloud_1

在设置过程中，您还需要注意连接设置。在本教程中，我们选择“任意位置”，但如果您申请特定位置，则需要将从您的 EMQX Cloud 部署中获得的 NAT 网关 IP 地址添加到白名单中。

clickhouse_cloud_2

然后您需要保存您的用户名和密码以备将来使用。

clickhouse_cloud_3

之后，您将获得一个正在运行的 Clickhouse 实例。单击“连接”以获取 Clickhouse Cloud 的实例连接地址。

clickhouse_cloud_4

单击“连接到 SQL 控制台”以创建数据库和表以与 EMQX Cloud 集成。

clickhouse_cloud_5

您可以参考以下 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)

clickhouse_cloud_6

在 EMQX Cloud 上创建 MQTT 服务​

在 EMQX Cloud 上创建专用的 MQTT 代理只需点击几下即可。

获取帐户​

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

从 EMQX Cloud 注册 页面开始，如果您是 EMQX Cloud 的新手，请单击“开始免费试用”以注册帐户。

emqx_cloud_sign_up

创建 MQTT 集群​

登录后，单击帐户菜单下的“云控制台”，您将能够看到创建新部署的绿色按钮。

emqx_cloud_create_1

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

选择专业版并选择“北弗吉尼亚州”区域，然后单击“立即创建”。只需几分钟，您将获得一个完全托管的 MQTT 代理

emqx_cloud_create_2

现在单击面板以转到集群视图。在此仪表板中，您将看到 MQTT 代理的概述。

emqx_cloud_overview

添加客户端凭据​

默认情况下，EMQX Cloud 不允许匿名连接，因此您需要添加客户端凭据，以便您可以使用 MQTT 客户端工具将数据发送到此代理。

单击左侧菜单上的“身份验证和 ACL”，然后单击子菜单中的“身份验证”。单击右侧的“添加”按钮，并为稍后的 MQTT 连接提供用户名和密码。这里我们将使用“emqx”和“xxxxxx”作为用户名和密码。

emqx_cloud_auth

单击“确认”，现在我们已准备好一个完全托管的 MQTT 代理。

启用 NAT 网关​

在开始设置 ClickHouse 集成之前，我们需要先启用 NAT 网关。默认情况下，MQTT 代理部署在私有 VPC 中，无法通过公共网络将数据发送到第三方系统。

返回“概述”页面并向下滚动到页面底部，您将看到 NAT 网关小部件。单击“订阅”按钮并按照说明操作。请注意，NAT 网关是一项增值服务，但也提供 14 天的免费试用。

emqx_cloud_nat_gateway

创建后，您将在小部件中找到公共 IP 地址。请注意，如果您在 ClickHouse Cloud 设置期间选择“从特定位置连接”，则需要将此 IP 地址添加到白名单中。

将 EMQX Cloud 与 ClickHouse Cloud 集成​

EMQX Cloud 数据集成 用于配置处理和响应 EMQX 消息流和设备事件的规则。数据集成不仅提供了一个清晰灵活的“可配置”架构解决方案，还简化了开发过程，提高了用户可用性，降低了业务系统与 EMQX Cloud 之间的耦合度。它还为定制 EMQX Cloud 的专有功能提供了卓越的基础设施。

emqx_cloud_data_integration

EMQX Cloud 提供了与流行数据系统超过 30 种的原生集成。ClickHouse 就是其中之一。

data_integration_clickhouse

创建 ClickHouse 资源​

单击左侧菜单上的“数据集成”，然后单击“查看所有资源”。您将在“数据持久化”部分找到 ClickHouse，或者您可以搜索 ClickHouse。

单击 ClickHouse 卡片以创建新资源。

  • 注意：为此资源添加注释。
  • 服务器地址：这是您的 ClickHouse Cloud 服务的地址，请记住不要忘记端口。
  • 数据库名称：我们在上述步骤中创建的“emqx”。
  • 用户：连接到您的 ClickHouse Cloud 服务的用户名。
  • 密钥：连接的密码。

data_integration_resource

创建新规则​

在创建资源期间，您会看到一个弹出窗口，点击“新建”将引导您进入规则创建页面。

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”主题的消息，并通过添加 client_id、topic 和时间戳信息来丰富 JSON 对象。

因此，您发送到主题的原始 JSON

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

data_integration_rule_1

您可以使用 SQL 测试来测试并查看结果。

data_integration_rule_2

现在点击“下一步”按钮。此步骤用于告诉 EMQX Cloud 如何将细化后的数据插入到您的 ClickHouse 数据库中。

添加响应操作​

如果您只有一个资源，则无需修改“资源”和“操作类型”。您只需要设置 SQL 模板。以下是本教程中使用的示例

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

data_integration_rule_action

这是一个将数据插入 Clickhouse 的模板，您可以看到此处使用了变量。

查看规则详情​

点击“确认”和“查看详情”。现在，一切都应该设置好了。您可以从规则详情页面查看数据集成是否正常工作。

data_integration_details

所有发送到“temp_hum/emqx”主题的 MQTT 消息都将持久保存到您的 ClickHouse Cloud 数据库中。

将数据保存到 ClickHouse​

我们将模拟温度和湿度数据，并通过 MQTT X 将这些数据报告给 EMQX Cloud，然后使用 EMQX Cloud 数据集成将数据保存到 ClickHouse Cloud 中。

work-flow

将 MQTT 消息发布到 EMQX Cloud​

您可以使用任何 MQTT 客户端或 SDK 发布消息。在本教程中，我们将使用MQTT X，这是 EMQ 提供的一个用户友好的 MQTT 客户端应用程序。

mqttx-overview

在 MQTTX 上点击“新建连接”并填写连接表单

  • 名称：连接名称。使用任何您想要的名称。
  • 主机：MQTT 代理连接地址。您可以从 EMQX Cloud 概览页面获取。
  • 端口：MQTT 代理连接端口。您可以从 EMQX Cloud 概览页面获取。
  • 用户名/密码：使用上面创建的凭据，在本教程中应该是“emqx”和“xxxxxx”。

mqttx-new

点击右上角的“连接”按钮，连接应该建立。

现在您可以使用此工具将消息发送到 MQTT 代理。输入：1. 设置有效负载格式为“JSON”。2. 设置主题：temp_hum/emqx（我们在规则中刚刚设置的主题）3. JSON 主体

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

点击右侧的发送按钮。您可以更改温度值并将更多数据发送到 MQTT 代理。

发送到 EMQX Cloud 的数据应由规则引擎处理并自动插入到 ClickHouse Cloud 中。

mqttx-publish

查看规则监控​

检查规则监控并将成功次数加一。

rule_monitor

检查持久化数据​

现在是时候查看 ClickHouse Cloud 上的数据了。理想情况下，您使用 MQTTX 发送的数据将进入 EMQX Cloud 并借助本地数据集成持久保存到 ClickHouse Cloud 的数据库中。

您可以连接到 ClickHouse Cloud 面板上的 SQL 控制台或使用任何客户端工具从您的 ClickHouse 获取数据。在本教程中，我们使用了 SQL 控制台。通过执行 SQL

SELECT * FROM emqx.temp_hum;

clickhouse_result

总结​

您没有编写任何代码，现在可以让 MQTT 数据从 EMQX Cloud 移动到 ClickHouse Cloud。使用 EMQX Cloud 和 ClickHouse Cloud，您无需管理基础设施，只需专注于编写将数据安全地存储在 ClickHouse Cloud 中的物联网应用程序。