别用 Python 写“滑动窗口”了！用 eKuiper (SQL) 榨干边缘网关的最后 1MB 内存 (附流处理规则)

架构师笔记

2026-03-11 10:34:00

#流处理 #eKuiper #SQL #边缘计算 #滑动窗口 #数据聚合 #LF_Edge

一、场景痛点：手写“滑动窗口”引发的内存惨案

上周在审查一个风机状态监测（CMS）项目的边缘端代码时，我看到了让所有老程序员血压飙升的一段 Python 代码：

业务需求：PLC 以 50Hz（每秒 50 条）的频率上报电机温度。要求：计算过去 5 秒的平均温度；如果平均温度 > 80℃，且持续 3 秒，则触发停机报警。
现状代码：开发工程师在 Python 里维护了一个全局 List。每次来数据就 list.append()，然后判断长度，超过 250 条就 list.pop(0)，最后用 sum()/len() 算平均值。
灾难现场：

CPU 飙升：当并发监控 20 台电机时，Python 疯狂的列表切片和垃圾回收 (GC) 把网关 CPU 吃到了 100%。
逻辑漏洞：如果 PLC 网络抖动，数据晚到了 2 秒，这个基于“到达时间”的死板数组会把错误的数据混在一起计算，导致误报警。
难以修改：甲方说把“5秒”改成“10秒”，工程师要在代码里把长度判断从 250 改成 500，极易出错。

架构师指令：停止在应用层用数组“造轮子”。

面对高频流式数据，正确的设计模式是引入 轻量级流处理引擎 (Stream Processing Engine)。我们采用 Linux 基金会开源的 eKuiper：一个只有 10MB 大小、内存占用不到 50MB，且能用 SQL 写实时逻辑的神器。

二、架构设计：SQL 定义数据流

eKuiper 的核心理念是：“把流动的数据当成一张不断变长的数据库表”。

你不需要写任何 Python 或 Go 代码，只需要写一条标准的 SQL 语句，剩下的内存管理、窗口划分、乱序处理全由引擎底层（基于 Go 语言）搞定。

Source (数据源)：MQTT 接收 PLC 发来的原始实时数据（高频）。
Rule (规则引擎)：在内存中执行 SQL 计算（如 Tumbling Window 翻滚窗口、Sliding Window 滑动窗口）。
Sink (目标端)：计算出结果后，将“报警指令”发回本地 MQTT 控制设备，或将“降维后的均值数据”发往云端。

数据流向拓扑：

PLC -> [本地 MQTT Broker] -> [eKuiper: 实时 SQL 窗口计算 (内存内)] -> [云端 / 报警动作]

三、核心实施步骤 (Copy & Paste)

假设你在网关上已经运行了 EMQX 和 eKuiper 容器。以下操作均可通过调用 eKuiper 的 REST API 瞬间完成热加载，无需重启进程。

1. 定义数据流 (Create Stream)

告诉 eKuiper 数据从哪里来，长什么样。我们要接入 MQTT 中 sensor/motor/# 的温度数据。

JSON

// POST http://localhost:9081/streams
{
  "sql": "CREATE STREAM motor_stream (motor_id STRING, temp FLOAT) WITH (DATASOURCE=\"sensor/motor/#\", FORMAT=\"json\")"
}

2. 编写流处理规则 (Create Rule) - 魔法发生的地方

我们要实现业务需求：按 motor_id 分组，计算过去 5 秒的平均温度，仅输出均值大于 80 的结果。

请看这句极其优雅的 SQL：

JSON

// POST http://localhost:9081/rules
{
  "id": "rule_high_temp_alarm",
  "sql": "SELECT motor_id, avg(temp) as avg_temp FROM motor_stream GROUP BY motor_id, TumblingWindow(ss, 5) HAVING avg(temp) > 80",
  "actions":[
    {
      "mqtt": {
        "server": "tcp://127.0.0.1:1883",
        "topic": "alarm/motor/{{.motor_id}}",
        "sendSingle": true
      }
    }
  ]
}

解析：TumblingWindow(ss, 5) 表示每 5 秒划分一个绝对的时间窗口。引擎会在内存中高效地累加数据，时间一到立即触发计算并清空窗口，绝不漏内存。

3. 进阶：检测“连续异常” (状态机)

如果甲方要求“连续 3 次平均温度超标才报警”怎么办？不需要写复杂的全局变量状态机。使用 eKuiper 的 LAG 算子或事件机制即可：

SQL

-- 检测当次温度 > 80，且前一次也 > 80 的情况SELECT temp FROM motor_stream WHERE temp >80ANDlag(temp) >80

四、踩坑复盘 (Red Flags)

1. 晚到数据 (Late Arrival) 的幽灵

现象：由于 4G 网络抖动，本该 10:01:00 到达的温度数据，10:01:05 才到达网关。eKuiper 把这条旧数据算进了当前的窗口里，导致均值计算错误。
对策
：区分 Event Time（数据产生时间） 和 Processing Time（数据到达时间）。

在 CREATE STREAM 时，必须指定时间戳字段 TIMESTAMP="ts"，并在规则中配置 Watermark (水位线) 容忍度（如容忍延迟 2 秒）。
codeSQLCREATE STREAM motor_stream (..., ts BIGINT) WITH (TIMESTAMP="ts")

2. JSON 数组解包的性能陷阱

坑：传感器发来的 Payload 经常是批量数组格式：{"data": [{"temp":80}, {"temp":81}]}。如果直接用 SQL 处理非常别扭。
对策：在 SELECT 语句前，使用 UNNEST 函数先将数组“展开”为多条独立的流记录，再进行窗口计算。

3. OOM 风险 (内存溢出)

警告：虽然 eKuiper 内存控制极好，但如果你手贱写了一个 SlidingWindow(hh, 24)（24小时滑动窗口），且数据频率是 1000Hz，引擎会试图在内存中缓存 8640 万条数据，网关必定死机。
规则：边缘端流处理的窗口时间严禁超过 1 小时。长周期的聚合计算必须扔给后端的时序数据库（如 TDengine）去做。

五、关联资源与选型

流处理引擎对 CPU 架构的依赖极小，真正实现了跨平台。