跳至主要内容
跳至主要内容
编辑此页

将 EMQX 与 ClickHouse 集成

连接 EMQX​

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

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

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

前提条件​

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

获取您的 ClickHouse Cloud 服务​

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

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

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

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

点击“连接到 SQL 控制台”以创建与 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 代理只需点击几下即可完成。

获取一个帐户​

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

在 EMQX Cloud 注册 页面上开始，然后点击开始免费注册一个帐户（如果您是 EMQX Cloud 的新手）。

创建一个 MQTT 集群​

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

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

选择 Pro 版本，选择 N.Virginial 区域，然后点击 立即创建。只需几分钟，您将获得一个完全托管的 MQTT 代理

现在点击面板进入集群视图。在此仪表板上，您将看到 MQTT 代理的概述。

添加客户端凭据​

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

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

点击“确认”，现在我们拥有一个完全托管的 MQTT 代理，已准备就绪。

启用 NAT 网关​

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

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

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

将 EMQX Cloud 与 ClickHouse Cloud 集成​

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

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

创建 ClickHouse 资源​

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

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

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

创建一个新规则​

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

EMQX 提供了一个强大的 规则引擎，可以在将原始 MQTT 消息发送到第三方系统之前转换和丰富 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 和 timestamp 信息来丰富 JSON 对象。

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

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

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

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

添加响应操作​

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

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

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

查看规则详细信息​

点击“确认”和“查看详细信息”。现在，一切都应该设置完毕。您可以从规则详细信息页面查看数据集成是否有效。

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

将数据保存到 ClickHouse​

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

将 MQTT 消息发布到 EMQX Cloud​

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

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

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

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

现在您可以使用此工具向 MQTT Broker 发送消息。输入

  1. 将 Payload 格式设置为“JSON”。
  2. 将主题设置为：temp_hum/emqx（我们在规则中设置的主题）
  3. JSON 正文
{"temp": 23.1, "hum": 0.68}

点击右侧的发送按钮。您可以更改温度值并向 MQTT Broker 发送更多数据。

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

查看规则监控​

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

检查已持久化的数据​

现在是时候查看 ClickHouse Cloud 上的数据了。理想情况下，您使用 MQTTX 发送的数据将进入 EMQX Cloud，并在原生数据集成功能的帮助下持久化到 ClickHouse Cloud 的数据库中。

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

SELECT * FROM emqx.temp_hum;

摘要​

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

    © . This site is unofficial and not affiliated with ClickHouse, Inc.