变更数据捕获案例

# 变更数据捕获(CDC)案例

# 案例概述

在 IoT 场景里，"变更数据捕获"指从连续的设备上报流里识别出真正发生变化的事件——指标突变跨越阈值、状态翻转、字段变化等。StreamSQL 的分析函数（lag / had_changed / changed_col / changed_cols / latest / acc_*）专为这类逐事件的状态计算设计：每条数据到达立刻求值，状态跨事件保留，可用在 SELECT 和 WHERE。

本案例集均为 IoT 场景，SQL 与输出均在 StreamSQL 实跑通过。

# 演示数据

下面"电流变化后超过阈值"三个案例共用这组模拟输入（两个设备交替上报电流）：

{"current": 300, "deviceId": 1, "ts": 1}
{"current": 400, "deviceId": 2, "ts": 2}
{"current": 200, "deviceId": 1, "ts": 3}
{"current": 200, "deviceId": 2, "ts": 4}
{"current": 500, "deviceId": 1, "ts": 5}
{"current": 200, "deviceId": 2, "ts": 6}
{"current": 400, "deviceId": 1, "ts": 7}
{"current": 600, "deviceId": 2, "ts": 8}

1
2
3
4
5
6
7
8

# 电流变化后超过阈值（lag）

单纯比较 current > 300 会在持续高电流时反复告警。真正想要的是"由不超过阈值变为超过阈值"的那一刻——隐含了"和上一次比"的判断，这正是 lag 的用途。

# 案例1：全局——整条流的上一次值

不区分设备，把整条流当作一个序列，记录"上一个值"。

SELECT current, ts
FROM stream
WHERE current > 300 AND lag(current) <= 300

1
2
3

lag(current) 无 OVER，状态跨所有事件共享。输出（电流从 ≤300 跳到 >300 的时刻）：

{"current": 400, "ts": 2}
{"current": 500, "ts": 5}
{"current": 400, "ts": 7}

1
2
3

注意

此规则检测整条流（混了所有设备）的变化。例如 ts5 的上一个值是 ts4 的 200（设备2），而非设备1 自己的上一次。要按设备区分，看案例2。

# 案例2：按设备分区——各设备各自的上一次值

PARTITION BY deviceId 让每个设备维护各自的"上一次值"，互不干扰。

SELECT current, deviceId, ts
FROM stream
WHERE current > 300 AND lag(current) OVER (PARTITION BY deviceId) < 300

1
2
3

输出（每个设备各自从 <300 跳到 >300 的时刻）：

{"current": 500, "deviceId": 1, "ts": 5}
{"current": 600, "deviceId": 2, "ts": 8}

1
2

# 案例3：只盯某一设备——OVER WHEN 限定 lag 范围

只关心 deviceId = 1。WHERE deviceId = 1 过滤，再用 OVER (WHEN deviceId = 1) 让 lag 只在该设备事件上更新状态，排除其它设备数据的影响。

SELECT current, deviceId, ts
FROM stream
WHERE current > 300 AND deviceId = 1
  AND lag(current) OVER (WHEN deviceId = 1) < 300

1
2
3
4

输出：

{"current": 500, "deviceId": 1, "ts": 5}

OVER (WHEN ...) 是分析函数的条件状态语义：只有满足条件的事件才更新状态，不满足时复用上次结果。除 lag 外，had_changed 等分析函数也都支持 OVER 子句。

# 其它变更检测能力

# 案例4：设备状态翻转告警（had_changed）

设备上报运行状态（running / fault / offline…）。状态一变就输出，不变时静默。

SELECT deviceId, status, ts
FROM stream
WHERE had_changed(true, status) == true

1
2
3

输入 running → running → fault → fault → running，输出首次及每次翻转的行。ignoreNull=true 让传感器掉线的 nil 不污染基准。

# 案例5：只上报变化的传感器字段（changed_cols）

边缘网关上行带宽有限，要求每条消息只携带变化的字段并加前缀，整条都不变时干脆不发。

SELECT changed_cols("c_", true, temperature, humidity) FROM stream

输入 (23,50) → (23,55) → (23,55) → (25,55)，输出：

{"c_temperature": 23, "c_humidity": 50}
{"c_humidity": 55}
（第三条全不变，不输出）
{"c_temperature": 25}

1
2
3
4

要检测/输出"任意一列变化"，把列名换成 "*"：had_changed(true, "*")、changed_cols("d_", true, "*")。

# 案例6：最新有效读数 + 累积能耗（latest + acc_*）

传感器偶尔掉线上报 nil。latest 始终返回最新非空值；acc_* 在规则生命周期内持续累积。

SELECT deviceId,
       latest(temperature)   AS last_temp,
       acc_sum(power)        AS total_power,
       acc_max(power)        AS peak_power,
       acc_count(*)          AS sample_cnt
FROM stream

1
2
3
4
5
6

# 对比：持续超阈值（窗口 + HAVING，不是变化检测）

注意

"持续 N 秒超阈值"不是变化检测，不要用 lag 或窗口上的 over(when)。主流引擎（Flink/Spark 等）都用窗口聚合 + HAVING。

要求母线电流持续 10 秒都大于 200A 才告警（中间有下跌不算）：滑动窗口聚合这 10 秒内全部事件（含 <200 的下跌），HAVING mn > 200 把出现过下跌的窗口过滤掉。

SELECT min(concurrency) AS mn, count(*) AS c
FROM stream
GROUP BY SlidingWindow(ss, 10)
HAVING mn > 200

1
2
3
4

注意 HAVING 引用的是 SELECT 别名 mn。

# CDC 能力地图

需求	用什么	例子
和上一次比 / 跨阈值	`lag` (+ `OVER PARTITION BY/WHEN`)	电流突变跨越阈值
是否变化	`had_changed` / `changed_col`	状态翻转、单字段跳变
变化的字段（动态列）	`changed_cols`	边缘带宽压缩
最新非空值（抗 nil）	`latest`	传感器掉线
生命周期累积	`acc_sum/max/min/count/avg`	总能耗、峰值
任一列变化	`had_changed(true, "*")`	整行变更审计
持续 N 秒超阈值	窗口聚合 + `HAVING`（非分析函数）	持续过载告警

# 📚 相关文档

分析函数 - 完整的分析函数语法与函数列表
SQL参考 - HAVING、窗口函数等完整 SQL 语法
IoT温度告警与指标聚合 - 窗口聚合类 IoT 案例

在 GitHub 上编辑此页

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