引言

MQTT 协议是一种专为物联网应用而设计的轻量级消息传输协议。它具有简单、开放、易于实现的特点，是物联网应用的理想选择。MQTT 数据以连续实时的方式进行传输，非常适合由流处理引擎进行处理。

EMQX 是一款大规模分布式物联网 MQTT Broker，能够高效、可靠地连接海量的物联网设备，并实时处理和分发消息和事件流数据。eKuiper 是一个开源的流处理引擎，可以对流数据进行过滤、转换和聚合等操作。

本文将向您展示如何使用 eKuiper 实时流处理引擎来处理来自 EMQX 的 MQTT 数据。

场景描述

假设我们有个 MQTT 主题 demo/sensor，用于在 EMQX 中接收温度和湿度数据。我们希望使用 eKuiper 订阅该主题，并用流处理技术对数据进行处理和分析。然后，我们可以根据分析结果，触发用户的 HTTP 服务，或者将结果保存到外部存储中。

EMQX

由于 EMQX 支持标准的 MQTT 协议，所以 eKuiper 可以连接到任何版本的 EMQX。在这里，我们使用 EMQX Cloud 提供的免费公共 MQTT Broker 进行测试：

eKuiper

eKuiper 可以部署在边缘或云端。我们可以使用 Docker 进行快速安装。

docker run -p 9081:9081 -d --name kuiper -e MQTT_SOURCE__DEFAULT__SERVER=tcp://broker.emqx.io:1883 lfedge/ekuiper:1.10.0

我们可以用这个命令拉取并运行 eKuiper 1.10.0 版本的 docker 镜像。我们将 REST API 端口设置为 9081，在本教程中，我们将使用 REST API 来管理 eKuiper。我们还通过环境变量把默认的 MQTT Broker 地址指向了 EMQX Cloud 集群。

如果您想使用其他方法安装 eKuiper，请查看安装指南。

EMQX ECP (EMQX Edge-to-Cloud Platform) 是专为云边协同而打造的高级 MQTT 平台。它提供了专业的 Web UI 让您可以方便地管理 eKuiper。在本教程中，您也可以使用 ECP 来管理 eKuiper。更多细节，请参考 ECP 文档。

配置 eKuiper 订阅 MQTT 数据流

MQTT 数据是一种无界的、连续的流式数据。在 eKuiper 中，我们使用流的概念来映射这种类型的数据。要处理 MQTT 数据，我们首先要创建一个流来描述数据。

我们用 eKuiper REST API 来创建一个流：

POST http://127.0.0.1:9081/streams
Content-Type: application/json

{
  "sql": "CREATE STREAM demoMqttStream (temperature FLOAT, humidity FLOAT) WITH (TYPE=\"mqtt\", DATASOURCE=\"demo/sensor\", FORMAT=\"json\", SHARED=\"true\")"
}

用 Postman 等 HTTP 客户端发送上面的请求，将创建一个名为 demoMqttStream 的流，它是 MQTT 类型的数据源。datasource 属性的值是 demo/sensor，表示订阅 MQTT 的 demo/sensor 主题。数据格式是 JSON。SHARED 选项表示这个流可以被所有规则共享。

注意：

我们运行 eKuiper docker 容器时，MQTT Broker 地址默认是 tcp://broker.emqx.io:1883。如果您用的是别的 MQTT Broker，请在安装时换成您的 Broker 地址。

如果您想改变 MQTT Broker 地址或其他 MQTT 连接参数，如认证相关配置，可以修改 data/mqtt_souce.yaml 文件里的设置。

您可以用 + 和 # 通配符订阅多个主题，在 datasource 属性里使用这些通配符。比如，demo/+ 是订阅所有以 demo/ 开头的主题。demo/# 是订阅所有以 demo/ 开头的主题和 demo/ 下的所有子主题。

流处理 MQTT 数据

在 eKuiper 中，我们用规则来定义流处理的工作流程。规则是 SQL 语句，它规定了数据处理的方式和处理后执行的动作。除了连续的数据处理，像 eKuiper 这样的流处理引擎还支持有状态处理。我们将演示两个流处理和有状态处理的例子。

有状态的报警规则

第一个流处理例子是监测温度和湿度数据，温度上升超过 0.5 或湿度上升超过 1 就触发报警。这要求处理引擎能够记住前一条数据的状态，并和当前数据比较。

假设我们有个 URL 为 http://yourhost/alert 的 HTTP webhook，用来接收报警数据。我们首先用下面的 HTTP 请求创建一个规则。

###
POST http://{
  
  {host}}/rules
Content-Type: application/json

{
  "id": "rule1",
  "sql": "SELECT temperature, humidity FROM demoMqttStream WHERE temperature - LAG(temperature) > 0.5 OR humidity - LAG(humidity) > 1",
  "actions": [{
    "rest": {
      "url": "http://yourhost/alert",
      "method": "post",
      "sendSingle": true
    }
  }]
}

上述请求创建了一个名为 rule1 的规则，该规则对应的 SQL 语句如下：

SELECT temperature, humidity 
FROM demoMqttStream 
WHERE 
  temperature - LAG(temperature) > 0.5 
  OR humidity - LAG(humidity) > 1

这个 SQL 从 demoMqttStream 里选出变化达到我们条件的温度和湿度数据。LAG 函数用来获取前一条数据。

actions 属性规定了规则触发后的动作。这里，我们用 rest 动作把数据发送到 http://yourhost/alert 。发送的是 SQL 筛选出的数据，以 JSON 格式发送。所以，发送的数据是这样的：

{
  "temperature": 25.5,
  "humidity": 60.5
}

测试规则

我们可以用 MQTTX 或者其他 MQTT 客户端来发布 MQTT 数据到 demo/sensor 主题。规则会处理这些数据。比如，我们发送以下数据到主题：

{"temperature": 25.5, "humidity": 60.5}
{"temperature": 26.1, "humidity": 62}
{"temperature": 25.9, "humidity": 62.1}
{"temperature": 26.5, "humidity": 62.3}

我们将在 HTTP 警报服务中收到下列数据：

{"temperature": 26.1, "humidity": 62}
{"temperature": 26.5, "humidity": 62.3}

这是因为只有第二和第四条消息，温度上升超 0.5 或湿度上升超 1。

时间窗口聚合规则

第二个例子是计算每分钟的平均温度和湿度，并把它发送回 EMQX。这涉及到一个经典的流处理概念，叫做时间窗口。我们可以用以下 HTTP 请求来创建一个规则。

###
POST http://{
  
  {host}}/rules
Content-Type: application/json

{
  "id": "rule2",
  "sql": "SELECT 
  trunc(avg(temperature), 2) as avg_temperature, trunc(avg(humidity), 2) as avg_humidity, window_end() as ts FROM demoMqttStream GROUP BY TumblingWindow(mi, 1)",
  "actions": [{
    "mqtt": {
      "server": "tcp://broker.emqx.io:1883",
      "topic": "result/aggregation",
      "sendSingle": true
    }
  }]
}

上述请求创建了一个名为 rule2 的规则，该规则对应的 SQL 语句如下：

SELECT 
  trunc(avg(temperature), 2) as avg_temperature, 
  trunc(avg(humidity), 2) as avg_humidity,
  window_end() as ts
FROM demoMqttStream
GROUP BY TumblingWindow(mi, 1)

这个 SQL 会选出每分钟的温度和湿度平均值。时间窗口在 GROUP BY 子句中用 TumblingWindow 定义。这种窗口类型把 MQTT 数据分成固定长度的窗口。在 SELECT 子句中，我们用聚合函数 avg 来计算时间窗口内温度和湿度的平均值。window_end() 函数用来获取时间窗口的结束时间，这样我们就能知道这些平均值对应的时间段。trunc 函数用来把平均值四舍五入到两位小数。

actions 属性规定了规则触发后的动作。这里，我们用 mqtt 动作发送数据到 EMQX 的 result/aggregation 主题。发送的是 SQL 筛选出的数据，以 JSON 格式发送。所以，发送到主题的数据是这样的：

{
  "avg_temperature": 25.5,
  "avg_humidity": 60.5,
  "ts": 1621419600000
}

测试规则

同样，我们可以用 MQTTX 或者其他 MQTT 客户端来发布 MQTT 数据到 demo/sensor 主题。规则会处理这些数据。比如，我们每 30 秒发送一条数据到主题，两分钟的数据如下所示：

{"temperature": 25.5, "humidity": 60.5}
{"temperature": 26.1, "humidity": 62}
{"temperature": 25.9, "humidity": 62.1}
{"temperature": 26.5, "humidity": 62.3}

我们将在 HTTP 警报服务中收到下列数据：

{"avg_temperature": 25.8, "avg_humidity": 61.25, "ts": 1621419600000}
{"avg_temperature": 26.2, "avg_humidity": 62.2, "ts": 1621419660000}

我们发送了两分钟的数据，所以得到了两个每分钟的平均值。

结语

在本教程中，我们学习了如何使用 eKuiper 处理 MQTT 数据。通过本教程，您能够：

• 通过订阅 EMQX MQTT Broker 主题接收 MQTT 数据
• 制定规则来处理 MQTT 数据
• 将处理后的数据反馈给 EMQX Broker

我们用两个示例展示了 eKuiper 对 MQTT 数据的流处理能力。eKuiper 强大的流处理能力可以应用于多种流式数据源。欢迎您探索 eKuiper 的各种功能，构建实时高效的 MQTT 数据处理通道。

版权声明： 本文为 EMQ 原创，转载请注明出处。
原文链接：https://www.emqx.com/zh/blog/mqtt-stream-processing-with-emqx-and-ekuiper