分析函数

分析函数（Analytic Functions）用于在无窗口的连续事件流上做跨事件的状态计算——比如"上一个值"、"是否发生变化"、"累积求和"。每条事件到达就立刻求值，状态跨事件保留。

何时用分析函数，何时用窗口聚合

• 变化检测 / 上下文回溯 / 生命周期累积（"和上一次比变了没"、"上一个值是多少"、"开服至今累计多少"）→ 用分析函数。
• 时间段统计 / 持续时长检测（"过去 10 秒的最小值"、"每分钟平均"）→ 用窗口聚合 + HAVING。 分析函数也能用在窗口查询里（对窗口输出求值，状态跨窗口保留，见在窗口查询中使用）；但纯粹的阈值/持续检测用窗口+HAVING 更直接。

概述

维度	分析函数	窗口聚合函数
触发时机	每条事件到达立刻求值	窗口触发时批量求值
状态	跨事件保留（流级状态机）	窗口内聚合，触发后清空
可用位置	SELECT、WHERE	SELECT、HAVING
典型场景	变化检测、lag、累积	计数、均值、最值、持续检测
是否需要 GROUP BY	不需要（纯分析查询走直连路径）	需要

OVER 子句

分析函数可带 OVER 子句控制状态的分区与更新条件：

func(args) OVER ( [PARTITION BY col[, col...]] [WHEN condition] )

• PARTITION BY：按字段分区，每个分区维护独立状态（如"每个设备各自的上一个值"）。
• WHEN condition：只有满足条件的事件才更新状态；不满足时复用上一次的结果（条件状态语义）。
• 不支持 ORDER BY / ROWS BETWEEN（那是 Flink 的窗口帧模型；streamsql 的分析函数是逐事件状态机）。

-- 无 OVER：全局一个状态
SELECT lag(temperature) AS prev FROM stream

-- PARTITION BY：每设备各自的状态
SELECT lag(temperature) OVER (PARTITION BY deviceId) AS prev FROM stream

-- WHEN：只在温度>20时更新状态
SELECT lag(temperature) OVER (WHEN temperature > 20) AS prev FROM stream

-- 组合：每设备、且仅在有效事件时更新
SELECT lag(temperature) OVER (PARTITION BY deviceId WHEN temperature > 20) AS prev FROM stream

求值时机与可用位置

分析函数在每条事件到达时立刻求值，且早于 WHERE——这样 WHERE 才能引用分析函数的结果（CDC 变化检测的核心）。

位置	是否支持	说明
SELECT	✅	投影输出，主用法
WHERE	✅	用分析函数过滤，如 WHERE lag(current) < 300
HAVING	❌	HAVING 只作用于窗口/聚合查询；分析函数不能进 HAVING
窗口查询内（SELECT）	✅	对窗口产出行求值，状态跨窗口保留（见在窗口查询中使用）

函数列表

lag —— 前一个值

返回当前行之前第 offset 个事件的值。

lag(field [, offset [, default [, ignoreNull]]])

参数	默认	说明
field	必填	目标字段
offset	1	往前回溯几步
default	nil	不足步数时的返回值
ignoreNull	false	true 时跳过 nil 值不计入历史

SELECT temperature, lag(temperature) AS prev FROM stream
SELECT lag(value, 2, -1, true) AS prev2 FROM stream   -- 往前2步, 默认-1, 跳过null

latest —— 最新非空值

返回指定字段的最新非空值，nil 不更新状态。

latest(field [, default])

SELECT latest(temperature) AS lt FROM stream

had_changed —— 是否发生变化

返回布尔值：与上次相比是否有变化（首次视为变化）。

had_changed(ignoreNull, field[, field...])

• ignoreNull=true：nil 不触发变化，且不污染基准。
• 支持多列：任一列变化即为 true。
• had_changed(ignoreNull, "*")：对整行所有列检测变化（* 展开）。

SELECT had_changed(true, temperature) AS chg FROM stream
SELECT ts FROM stream WHERE had_changed(true, status) == true
SELECT * FROM stream WHERE had_changed(true, "*") == true   -- 任一列变化

changed_col —— 变化的列值（单列标量）

变化时返回新值，未变化返回 nil。可用于 WHERE 和 SELECT。

changed_col(ignoreNull, field)

SELECT changed_col(true, temperature) AS chg FROM stream

changed_cols —— 多列变化值（仅 SELECT，动态列）

对多列检测变化，只输出发生变化的列，输出列名 = prefix + 原列名。支持 "*" 对整行检测。

changed_cols(prefix, ignoreNull, field[, field...])

-- temperature 变化 → 输出 c_temperature；humidity 变化 → 输出 c_humidity
SELECT changed_cols("c_", true, temperature, humidity) FROM stream
SELECT changed_cols("d_", true, "*") FROM stream   -- 整行各列

omitEmpty

当 changed_cols 是查询的唯一输出且本次无任何变化时，整行不输出（事件压缩）。配合普通字段时，普通字段照常输出，只有变化列按需出现。

acc_* —— 生命周期累积

规则整个生命周期内持续累积，不随窗口重置。

函数	说明
acc_sum(field)	累积求和
acc_max(field)	累积最大值
acc_min(field)	累积最小值
acc_count(field)	累积计数（非 nil）
acc_avg(field)	累积平均值

SELECT acc_sum(power) AS total, acc_max(power) AS peak, acc_count(*) AS cnt FROM stream

与聚合函数 sum 的区别：sum 是窗口内聚合，每个窗口结束清零；acc_sum 跨整个生命周期累加、不重置，走逐事件状态机（可用 EmitSync）。

-- sum：每 5 秒这批各求一次和，窗口结束清零
SELECT deviceId, sum(current) FROM stream GROUP BY deviceId, TumblingWindow('5s')
-- acc_sum：每个设备从开机至今持续累加，不重置
SELECT acc_sum(current) OVER (PARTITION BY deviceId) AS total FROM stream

acc_* 的分区淘汰会静默归零

acc_* 不随窗口重置，但若该分区被 PARTITION 上限淘汰（见下文"误区 3"），累计值会静默从 0 重新开始——设备断流一段时间后回来，计数不会接着涨，而是像新设备。对累计准确性敏感时，把上限调到高于活跃分区数。

条件累计（开始点 / 重置点）

acc_* 可额外接受两个布尔表达式：第一个为开始点（满足才开始累计，之后持续），第二个为重置点（满足则归零并停止，直到再次满足开始点）。

acc_count(expr, startExpr, resetExpr)

-- 仅在 a>1 后开始计数，a<0 时归零
SELECT acc_count(a, a > 1, a < 0) AS c FROM stream
-- 输入 a: 1,2,1,3,-1,1 → c: 0,1,2,3,0,0

在窗口查询中使用分析函数

分析函数可用于窗口查询：此时它在窗口产出行上求值，状态跨窗口保留（不随窗口清零），等价于"对每个窗口的输出做变化检测 / 回溯 / 累计"。PARTITION 默认取 GROUP BY 键——每个分组各自跨窗口维护状态，互不串扰。

分析函数的参数可以是聚合函数（内联写法），例如对窗口聚合结果做变化检测：

-- 每两个事件一个计数窗口求平均；输出平均温度发生变化的窗口
SELECT changed_cols("t", true, avg(temperature))
FROM stream
GROUP BY CountingWindow(2)
-- 输入温度 23,23,23,25,25,25,25,25 → 窗口均值 23,24,25,25
-- 输出 {tavg:23} {tavg:24} {tavg:25}（最后一个 25→25 无变化被抑制）

-- 跨窗口累计求和：每窗口均值累加
SELECT acc_sum(avg(temperature)) AS total FROM stream GROUP BY CountingWindow(2)

别名形式

也可以先给聚合起别名，再用分析函数引用：SELECT avg(temperature) AS a, changed_col(true, a) AS chg ... GROUP BY 窗口。内联与别名两种写法等价。

参数需引用窗口输出字段

窗口查询里分析函数的参数必须是聚合函数或 GROUP BY 字段——它们出现在窗口产出行里。引用裸原始列（既未聚合、也非分组键）取不到值。

详细案例

案例 1：CDC 变化检测——电流突变超阈值

场景：每个设备的电流从低变高、跨越 300A 阈值时输出。要求"当前 > 300 且 上一次 < 300"。

数据：

{"current": 300, "deviceId": 1, "ts": 1}
{"current": 200, "deviceId": 1, "ts": 2}
{"current": 500, "deviceId": 1, "ts": 3}   // 1号: 200→500 跨阈值
{"current": 200, "deviceId": 2, "ts": 4}
{"current": 600, "deviceId": 2, "ts": 5}   // 2号: 200→600 跨阈值

SQL（lag 配 PARTITION BY 进 WHERE）：

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

输出：

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

要点：PARTITION BY deviceId 让每个设备各自维护"上一次电流"；lag 在 WHERE 之前求值，所以 WHERE 能引用它。

案例 2：状态变化过滤——只输出变化的行

场景：温度不变时不输出，只在变化时输出。

SQL：

SELECT ts, temperature
FROM stream
WHERE had_changed(true, temperature) == true

输入/输出：

{ts:1, temperature:23} → 输出（首次视为变化）
{ts:2, temperature:23} → 不输出（未变化）
{ts:3, temperature:25} → 输出（变化）

案例 3：事件压缩——只发送变化字段

场景：上行带宽有限，每条消息只携带发生变化的字段，并加前缀 c_。

SQL：

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

输入/输出：

{temperature:23, humidity:50} → {c_temperature:23, c_humidity:50}   // 首次全变
{temperature:23, humidity:55} → {c_humidity:55}                     // 只 humidity 变
{temperature:23, humidity:55} → (无输出)                            // 都没变，整行抑制

案例 4：设备级上一次读数

场景：报表里同时展示当前温度和每个设备的上一次温度。

SELECT deviceId, temperature, lag(temperature) OVER (PARTITION BY deviceId) AS prev_temp
FROM stream

案例 5：累积统计

场景：统计开机以来的总能耗、峰值、采样次数。

SELECT acc_sum(power) AS total, acc_max(power) AS peak, acc_count(*) AS cnt, acc_avg(power) AS avg_power
FROM stream

案例 6：条件状态——只在有效事件时更新

场景：设备偶尔上报异常值（temperature ≤ 0 为无效）。lag 应忽略无效事件，只在有效事件时更新基准。

SELECT lag(temperature) OVER (WHEN temperature > 0) AS prev_valid FROM stream

不满足 temperature > 0 的事件不更新状态，prev_valid 复用上一次有效值。

案例 7：持续超阈值检测（窗口 + HAVING）

注意

"持续 N 秒超阈值"不要用分析函数或窗口上的 over(when)——主流引擎（Flink/Spark 等）都用窗口聚合 + HAVING。

场景：电流持续 10 秒都大于 200A 才告警。

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

原理：滑动窗口聚合这 10 秒内的全部事件（含 <200 的下跌），HAVING mn > 200 把出现过下跌的窗口过滤掉——留下来的就是"全程 >200"的窗口。没有任何窗口满足 HAVING 时，不输出（每个窗口聚合后逐个过 HAVING，全被过滤则该周期无输出）。

HAVING 引用别名

streamsql 的 HAVING 引用 SELECT 里的别名（mn），不能复述聚合函数（HAVING min(concurrency) > 200 不生效）。

案例 8：窗口内变化检测——平均值变化才上报

场景：上行带宽有限，每两个采样求一次平均，只在平均值发生变化时上报（均值稳定时不发）。

SELECT changed_cols("t", true, avg(temperature)) FROM stream GROUP BY CountingWindow(2)

输入/输出（温度序列 23,23,23,25,25,25,25,25）：

窗口均值：23, 24, 25, 25
输出：{tavg:23} {tavg:24} {tavg:25}   ← 最后一个 25→25 无变化，被抑制

分析函数在窗口产出后对均值求值，状态跨窗口保留，与上一个窗口的均值比较。

案例 9：设备级跨窗口变化——各设备独立追踪

场景：多设备同流，每个设备分别追踪自己窗口均值的变化，互不串扰。

SELECT deviceId, changed_col(true, avg(temp)) AS chg
FROM stream
GROUP BY deviceId, CountingWindow(2)

输入/输出（A: 10,20,30,40；B: 5,5）：

A 窗口均值：15（首→变化）、35（变化）   → {deviceId:A, chg:15} {deviceId:A, chg:35}
B 窗口均值：5（首→变化）                → {deviceId:B, chg:5}

窗口查询里分析函数默认按 GROUP BY 键分区，每个设备各自跨窗口维护状态。

案例 10：条件累计——过载脉冲内计数

场景：统计每次过载脉冲（电流 >200A）内的事件数，电流跌回（<10A）时归零，进入下一次脉冲重新计。

SELECT acc_count(current, current > 200, current < 10) AS c FROM stream

输入/输出（current: 50, 250, 300, 5, 280）：

50  → 0   （未进入过载，不计）
250 → 1   （current>200，开始计数）
300 → 2
5   → 0   （current<10，归零）
280 → 1   （再次过载，重新计数）

条件表达式用比较运算符

acc_* 的开始/重置条件是布尔表达式，用 > < >= == 等比较运算符（走表达式引擎）。注意 SQL 的 = 在分析函数参数里不作为相等判断，字符串相等请用 ==。

常见误区

误区 1：over(when) 挂窗口 vs 挂分析函数

-- ✅ over 挂在分析函数 lag 上：合法，CDC 用法
WHERE lag(current) OVER (PARTITION BY deviceId) < 300

-- ❌ over 挂在 GROUP BY 窗口上：不支持，会报错
GROUP BY SlidingWindow(ss, 10) OVER (WHEN concurrency > 200)

窗口上的阈值/持续检测用 HAVING（见案例 7）。

误区 2：分析函数不能进 HAVING；窗口内分析函数别引用裸列

分析函数不能进 HAVING（HAVING 只作用于窗口/聚合查询）。

分析函数可以出现在窗口查询的 SELECT 里（对窗口输出求值，见在窗口查询中使用），但其参数必须是聚合函数或 GROUP BY 字段。下面这种引用裸原始列的写法取不到值（temperature 不在窗口产出行里）：

-- ⚠️ 窗口查询里 lag(temperature)：temperature 未聚合，不在窗口输出中 → 返回 nil
SELECT lag(temperature) FROM stream GROUP BY TumblingWindow('5s')

要对窗口内的温度做回溯，应聚合后再分析：lag(avg(temperature)) 或先 avg(temperature) AS a 再 lag(a)。

误区 3：partition by 的高基数内存

PARTITION BY 的分区数会随不同键值增长（如海量 deviceId）。streamsql 对每个分析函数字段设分区上限（默认 10000），超出按 LRU 淘汰最久未使用的分区，防止内存无限增长。被淘汰的分区再次出现时会重置为初始状态：lag 重新返回 nil、had_changed/changed_col 当成首事件、acc_* 累计值静默归零。

默认 10000 覆盖单边缘节点的常见设备规模（每分区约 150B，changed_cols/had_changed(*) 宽行可达数百字节，1 万分区约几 MB，仍在边缘定位内）。设备上万且内存充裕的聚合网关，用 WithAnalyticMaxPartitions 调高：

ssql := streamsql.New(streamsql.WithAnalyticMaxPartitions(50000))

何时调高

仅当分区键确实是高基数（如单节点带数万设备）且内存允许时调高。误把近唯一字段（如时间戳、事件 id）当 PARTITION 键会让分区数暴涨——这种情况下调高只是推迟问题，应换低基数键。