IoT 温度告警与指标聚合

# IoT 温度告警与指标聚合案例

# 案例概述

物联网温度监控场景里，往往同时存在两类需求：一是单点越限即时告警（每来一条读数就要判断并下发），二是按设备滚动聚合指标（每分钟统计均值/极值，用于仪表盘与异常检测）。两类需求对应 StreamSQL 的两种工作模式：非聚合的同步转换（EmitSync）与基于窗口的异步聚合（Emit + AddSink）。

本案例围绕一个统一的温度读数流 stream，串联五个子场景：

子场景	模式	窗口	核心能力
A. 单点温度告警	非聚合（同步）	无	`EmitSync` + `CASE WHEN` 即时转换/过滤
B. 滚动 1 分钟设备指标	事件时间聚合	`TumblingWindow('1m')`	Watermark、乱序容忍、空闲推进
C. 高基数设备异常检测	处理时间聚合	`CountingWindow(50)`	`STATETTL` 死 key 清理
D. 确定性测试	触发器	`TumblingWindow('5s')`	`TriggerWindow()` 立即冲窗口
E. 全局窗口条件驱动告警	处理时间聚合	`GLOBAL WINDOW TRIGGER WHEN`	`FIRE_AND_PURGE`、聚合值/复合条件触发

如何运行这些 SQL

本文每个场景只给 SQL + 输入 + 输出；运行方式见案例集锦 · 如何运行案例 SQL。非聚合查询（场景 A）用 EmitSync 同步拿单条结果；聚合/窗口查询（场景 B–E）用 Emit + AddSink 按批回调。

# 场景 A：单点温度告警（非聚合，EmitSync）

# 业务目标

每条温度读数到达后立即完成：华氏度换算、分级标注（OK/WARNING/CRITICAL），并只下发温度大于 30 的读数。要求一进一出、同步可阻塞调用，调用方直接拿到结果。

# SQL

SELECT deviceId,
       temperature,
       temperature * 1.8 + 32 AS temp_f,
       CASE WHEN temperature > 35 THEN 'CRITICAL'
            WHEN temperature > 30 THEN 'WARNING'
            ELSE 'OK' END AS level
FROM stream
WHERE temperature > 30

1
2
3
4
5
6
7
8

# 输入与输出

输入 4 条读数（28 不满足 temperature > 30，被过滤）：

{"deviceId": "dev-01", "temperature": 28.0}
{"deviceId": "dev-02", "temperature": 32.0}
{"deviceId": "dev-03", "temperature": 38.0}
{"deviceId": "dev-04", "temperature": 42.0}

1
2
3
4

用 EmitSync 逐条同步拿结果，输出（28 被过滤）：

[filtered] dev-01
[alert]   {deviceId:dev-02 temperature:32 temp_f:89.6  level:WARNING}
[alert]   {deviceId:dev-03 temperature:38 temp_f:100.4 level:CRITICAL}
[alert]   {deviceId:dev-04 temperature:42 temp_f:107.6 level:CRITICAL}

1
2
3
4

# 行为说明

EmitSync 对非聚合查询同步返回单条结果；不命中 WHERE 的行返回 nil（不是 error），可用于区分"被过滤"与"出错"。
对聚合查询调用 EmitSync 会返回错误：synchronous mode only supports non-aggregation queries。
CASE WHEN ... THEN ... ELSE ... END 既可做分级标注，也可配合 WHERE 做条件过滤。

# 场景 B：滚动 1 分钟设备指标（事件时间 + Watermark）

# 业务目标

按设备维度，每 1 分钟输出一次窗口内的样本数、平均/最高/最低温度。要求：

以读数自带的时间戳为准（事件时间），而非处理时间；
容忍网络抖动导致的 5 秒乱序；
设备掉线后窗口仍能最终关闭，不留死数据。

# SQL

SELECT deviceId,
       COUNT(*) AS samples,
       AVG(temperature) AS avg_t,
       MAX(temperature) AS max_t,
       MIN(temperature) AS min_t,
       window_start()  AS w_start
FROM stream
WHERE temperature IS NOT NULL
GROUP BY deviceId, TumblingWindow('1m')
WITH (TIMESTAMP='ts', TIMEUNIT='ms', MAXOUTOFORDERNESS='5s', IDLETIMEOUT='10s')

1
2
3
4
5
6
7
8
9
10

WITH 各参数含义：

参数	含义
`TIMESTAMP='ts'`	用字段 `ts` 作为事件时间
`TIMEUNIT='ms'`	`ts` 的单位是毫秒（默认 `s`）
`MAXOUTOFORDERNESS='5s'`	Watermark = `max(event_time) - 5s`，容忍 5 秒乱序
`IDLETIMEOUT='10s'`	数据源空闲 10 秒后，按处理时间推进 Watermark，避免掉线设备卡死窗口

# 输入与输出

前 5 条全部落入第一个 1 分钟窗口 [0, 60000ms)，部分故意乱序（在 5s 容忍内）；最后一条 ts=66000 把 Watermark 推过 60000，触发第一个窗口输出：

{"deviceId": "dev-01", "ts": 1000,  "temperature": 31.0}
{"deviceId": "dev-01", "ts": 3000,  "temperature": 33.0}
{"deviceId": "dev-02", "ts": 2000,  "temperature": 29.0}
{"deviceId": "dev-01", "ts": 1500,  "temperature": 30.0}
{"deviceId": "dev-02", "ts": 4000,  "temperature": 27.0}
{"deviceId": "dev-03", "ts": 66000, "temperature": 25.0}

1
2
3
4
5
6

输出（窗口关闭时按批回调，每个设备一行）：

[window] {deviceId:dev-01 samples:3 avg_t:31.33 max_t:33 min_t:30 w_start:0}
[window] {deviceId:dev-02 samples:2 avg_t:28    max_t:29 min_t:27 w_start:0}

1
2

# 行为说明

输出不是逐条，而是窗口关闭时按批回调；同一窗口内每个 deviceId 一行。
dev-01@1500 晚于 dev-01@3000 到达，但因 MAXOUTOFORDERNESS='5s' 仍被正确纳入窗口。
若没有任何后续数据，IDLETIMEOUT='10s' 会在静默 10 秒后用处理时间推进 Watermark，关闭窗口。
window_start() 返回纳秒时间戳；如需毫秒展示请自行除以 1e6。

# 场景 C：高基数设备异常检测（CountingWindow + State TTL）

# 业务目标

设备规模很大（数万到数十万），其中一部分设备只上报几次就离线。要求：每凑满 50 条读数输出一次该设备的统计（均值 + 标准差），同时不能让"永不凑满 50 条"的设备把状态撑爆内存。

# SQL

SELECT deviceId,
       COUNT(*)           AS cnt,
       AVG(temperature)   AS avg_t,
       STDDEV(temperature) AS stddev_t
FROM stream
GROUP BY deviceId, CountingWindow(50)
WITH (STATETTL='1h')

1
2
3
4
5
6
7

要点：

CountingWindow(50) 按条数触发，与时间无关，只支持处理时间（不解析 TIMESTAMP/TIMEUNIT）。
STATETTL='1h'：超过 1 小时未再次触发的 group key 会被后台清理（lazy，Start goroutine 中执行）。
默认 STATETTL=0 表示禁用（与 Flink 对齐）。也就是说，不开 STATETTL 时，只上报过几条的设备会把计数器状态一直留在内存里——这就是"高基数死 key"问题。

# 输入与输出

dev-A 连续上报 50 条（温度在 20.0~24.0 间规律波动）凑满即触发；dev-B 只上报 3 条就消失，永不触发，靠 STATETTL 回收：

{"deviceId": "dev-A", "temperature": 20.0}   // ×50，温度 20.0..24.0
{"deviceId": "dev-B", "temperature": 99.0}   // ×3 后消失

1
2

输出（只有凑满的 dev-A）：

[burst] {deviceId:dev-A cnt:50 avg_t:22 stddev_t:1.414}

# 行为说明

dev-A 凑满 50 条立即输出，cnt=50，avg_t=22.0（20..24 均值），stddev_t≈1.414（总体标准差）。
dev-B 永远不会出现在输出里，但它的状态会保留：没有 STATETTL 时，大量这样的设备会把内存撑爆；开启 STATETTL='1h' 后，超过 1 小时未再触发的 key 被清理。
STDDEV 为总体标准差；如需样本标准差用 STDDEVS。

# 场景 D（加餐）：用 TriggerWindow 写确定性测试

# 业务目标

时间窗口在测试里很难用——你不想真的 time.Sleep(5*time.Second) 等窗口关闭。TriggerWindow() 提供了一个手动触发当前窗口立即输出的能力，让测试既快又确定。

# SQL

SELECT deviceId, COUNT(*) AS cnt
FROM stream
GROUP BY deviceId, TumblingWindow('5s')

1
2
3

# 输入与输出

只 Emit 一条，然后用 ssql.TriggerWindow() 立即冲窗口，不必等满 5 秒：

{"deviceId": "dev-X"}

[triggered] {deviceId:dev-X cnt:1}

# 行为说明

TriggerWindow() 对时间窗口立即冲窗口并输出；对 CountingWindow 是 no-op（按条数触发，与时间无关）；对非窗口查询也是安全 no-op。
因此它可以放心写进测试 teardown：无论被测 SQL 是哪种形态，都不会 panic。

# 场景 E：全局窗口条件驱动告警（GLOBAL WINDOW + TRIGGER WHEN）

# 业务目标

场景 A 是"每来一条立即判断单点温度"；但有些告警要看累积态势——"任一时刻该设备的运行态最高温越过阈值才告警"，或"攒满 N 条读数结算一次"。这类"条件命中才输出"的需求没有固定时间边界，也未必按条数，对应 StreamSQL 的全局窗口（对齐 Flink GlobalWindows + Trigger）：无内置边界、默认永不自动触发，由 TRIGGER WHEN 谓词驱动——每条数据更新运行态聚合，命中谓词就输出当前结果并清空该组状态（FIRE_AND_PURGE）。

# SQL

SELECT deviceId,
       MAX(temperature) AS max_t,
       COUNT(*)         AS samples
FROM stream
GROUP BY deviceId, GLOBAL WINDOW TRIGGER WHEN MAX(temperature) > 50

1
2
3
4
5

# 输入与输出

dev-01 逐步升温，55 越线立即触发（max_t=55、samples=3）并清零；随后的 48 进入新组，未越线：

{"deviceId": "dev-01", "temperature": 40.0}
{"deviceId": "dev-01", "temperature": 45.0}
{"deviceId": "dev-01", "temperature": 55.0}
{"deviceId": "dev-01", "temperature": 48.0}

1
2
3
4

[global] {deviceId:dev-01 max_t:55 samples:3}

# 行为说明

40、45 到达时运行态 max=45 ≤ 50，不触发；55 到达时 max=55 > 50 命中谓词，输出本组当前聚合（max_t=55、samples=3）并清空该组状态。
随后的 48 进入一个全新组，从 0 重新累积；max=48 ≤ 50 不再触发。若这时再来一条 52，会立即再次触发（max_t=52、samples=1）——这正是 FIRE_AND_PURGE "触发即清零、下批重来"的语义。
与场景 A 的关键区别：场景 A 每条判断单点值（temperature > 30）；这里判断的是累积最高温（MAX(temperature) > 50），只有全局窗口能做到。

# 📚 相关文档

如何运行案例 SQL —— 统一的 Go 运行样板
变更数据捕获 —— 分析函数做变化检测
SQL 参考 —— 完整 SQL 语法与函数列表
API 参考 —— EmitSync / Emit / AddSink / TriggerWindow 详解

在 GitHub 上编辑此页

上次更新: 2026/07/11, 04:40:00