提高数据集成稳定性：EMQX Platform 端到端规则调试指南

EMQX Team

2024-8-28

EMQX

EMQX 规则引擎介绍
为什么需要规则测试？
规则测试功能使用教程
结语

自 5.7.0 版本起，EMQX 支持了 SQL 调试，并支持在数据集成全流程中进行规则调试，使用户能够在开发阶段就全面验证和优化规则，确保它们在生产环境中的稳定高效运行。

点击此处下载 EMQX 最新版本：https://www.emqx.com/zh/try?tab=self-managed

本文将为您提供 EMQX 数据集成规则的调试指南，通过调试步骤的详细介绍，帮助您充分了解并利用这一强大的功能。

EMQX 规则引擎介绍

EMQX 规则引擎是一个基于 SQL 的数据处理组件，借助数据集成，用户无需编写代码即可完成物联网数据的提取、过滤、转换、存储和处理任务。

EMQX 规则引擎原理图

规则引擎的工作原理：

数据源：通过 SQL 中的 FROM 子句指定。
数据处理：通过 SQL 语句和函数进行描述。
结果输出：通过动作来处理输出结果，比如将其存储到数据库或重新发布到 MQTT 主题中。

关键动作：

将消息重新发布到 MQTT 主题
输出到控制台
发送到外部数据系统，如 Kafka、MySQL、PostgreSQL 等。

本指南将重点介绍如何在 EMQX 数据集成中调试这些规则，以确保它们在生产环境中能够按预期工作。

为什么需要规则测试？

目前，在 EMQX 中创建直接可用的规则可能会面临一些挑战，主要体现在以下几个限制：

仅限于 SQL 测试：当前只能通过模拟数据输入来测试 SQL，这虽然有助于用户调整 SQL 语法以实现目标，但无法全面验证规则的整体效果。
动作测试：动作只能在 Sink 资源的生产环境中观察，限制了提前充分测试和验证动作的可能性。每次调整都需要切换不同的系统来检查结果，增加了操作的复杂性。
模拟数据限制：模拟数据通常有限，难以全面反映 MQTT 属性和事件的真实情况，导致测试结果可能无法覆盖所有用户场景。

通过仪表板提供全面反馈并跟踪整个数据集成生命周期中的日志，规则测试可以有效解决上述限制，帮助用户更好地识别和解决问题。规则测试的主要优势包括：

端到端验证：可以使用真实的数据源和动作进行测试，确保从输入到输出的每个数据流环节都按预期运行。
更好的问题检测：在开发和规则编辑阶段，用户界面提供了直观的工具，帮助快速识别和解决潜在问题，从而减少生产环境中的故障发生率。
提高开发效率：测试功能能够缩短开发和测试周期，使规则的部署更加高效。
保障系统稳定性：通过预先测试和验证规则，降低了生产环境中出现未预见问题的风险。

规则测试功能使用教程

安装 EMQX Enterprise

推荐下载：EMQX Enterprise - 此版本提供丰富的数据集成功能，支持 Kafka、RabbitMQ、MySQL、PostgreSQL、InfluxDB、TimescaleDB 等常用的关系型数据库、时序数据库和流处理中间件。

您也可以使用以下 Docker 命令安装：

docker run -d --name emqx-enterprise -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx-enterprise:5.7.1

安装完成后，在浏览器中打开 <http://<your-host-address>>:18083，输入默认的用户名和密码即可登录 Dashboard。

登录后，依次点击数据集成 → 规则 → 创建，进入规则引擎创建页面。页面的上半部分用于配置 SQL、数据源和 Sink，而下半部分则是进行规则测试的区域。接下来，我们将通过两个简单的例子演示如何创建和启用规则测试。

EMQX Dashboard

创建规则

在完成 EMQX 安装后，我们通过一个具体示例来展示规则引擎的使用。该场景模拟车辆数据上报，并在车速超过 120 公里/小时时，将相关数据（包括速度和地理位置）发布到 HTTP 服务。车辆的 ID 和 MQTT 连接信息将存储在 PostgreSQL 数据库中。

模拟 MQTT 有效载荷进行测试：首先，模拟一条 MQTT 有效载荷来模拟车辆数据上报。

{
  "vehicle_id": "VH-958-XYZ",
  "speed": 72,
  "location": {
    "latitude": 40.7128,
    "longitude": -74.0060
  },
  "safety_features": {
    "airbag_deployed": false,
    "abs_status": "active"
  },
  "timestamp": "2024-07-11T15:45:00Z"
}

创建 SQL 规则：在 SQL 编辑器中创建规则，监听 devices/# 主题，并设置条件为车速超过 120 公里/小时时触发。

SELECT
  username,
  clientid,
  payload.vehicle_id as vehicle_id,
  payload.speed as speed,
  payload.location.latitude as latitude,
  payload.location.longitude as longitude,
  payload.safety_features.airbag_deployed as airbag_deployed,
  payload.safety_features.abs_status as abs_status,
  timestamp
FROM
  "devices/#"
WHERE payload.speed > '120'

测试和调试 SQL：编写完 SQL 后，在 SQL 选项卡中点击“开始测试”按钮，调试 SQL 语法。您可以输入模拟数据进行测试，调整和优化 SQL 语句。关于编写 SQL 的详细指南，请参考 EMQX 文档。测试完成后，我们将设置 HTTP 和 PostgreSQL 环境，为该规则添加两个动作。

设置 HTTP 服务

首先，我们使用 Node.js 创建一个简单的 HTTP 服务，用于接收和显示来自 EMQX 的数据。

const express = require("express");
const app = express();
app.use(express.json());

app.post("/speed", (req, res) => {
  const { speed, latitude, longitude } = req.body;
  console.log(`Received data: Speed is ${speed} km/h at coordinates (${latitude}, ${longitude}).`);
  res.status(200. send("Data received successfully!");
});

const PORT = 3000;
app.listen(PORT, () => {
  console.log(`Server is running on port ${PORT}`);
});

在规则创建页面中，选择添加动作，创建 HTTP 连接器，并配置 HTTP 请求主体，将从 SQL 提取的数据发送到上述 HTTP 服务。

编辑动作

配置 PostgreSQL 数据库

接下来，我们使用 Docker 快速部署 PostgreSQL 数据库，并使用 Postico 或其他 GUI 客户端管理数据库。

services:
  postgres:
    image: postgres
    environment:
      POSTGRES_DB: emqx
      POSTGRES_USER: admin
      POSTGRES_PASSWORD: public
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

部署完成后，我们需要创建数据库表，然后添加动作，以便将数据存储到数据库中。

CREATE TABLE car_infos (
  id SERIAL PRIMARY KEY,
  username VARCHAR(255),
  clientid VARCHAR(255),
  timestamp TIMESTAMP,
  vehicle_id VARCHAR(255),
  speed INT
);

数据库表创建完成后，回到规则创建页面中，再次点击添加动作，创建 PostgreSQL 连接器。使用该连接器创建一个动作，并在 SQL 模板中输入如下的 INSERT 语句，以确保当规则触发时，过滤后的数据能够保存到数据库中：

INSERT INTO car_infos (
  username,
  clientid,
  timestamp,
  vehicle_id,
  speed
) VALUES (
  ${username},
  ${clientid},
  TO_TIMESTAMP(${timestamp}::bigint / 1000),
  ${vehicle_id},
  ${speed}
);

编辑动作