RuleGo
目录

模式识别（MATCH_RECOGNIZE / CEP）

# 模式识别（MATCH_RECOGNIZE / CEP）

模式识别（CEP，Complex Event Processing）用于在事件流上识别按特定顺序出现的事件序列——比如"连续 3 次越限"、"先升后降"、"开机→运行→停机"。它用 SQL:2016 标准的 MATCH_RECOGNIZE 子句描述模式，对齐 Flink SQL（不是 FlinkCEP Java API）。

何时用模式识别，何时用分析函数 / 窗口

  • 多事件的顺序/序列模式（"连续 N 次超阈值"、"A 之后接 B"、"V 型反转"、"工作流 Start→End"）→ 用模式识别。
  • 相邻两事件的比较（"和上一次比变了没"、"上一个值是多少"）→ 用分析函数（lag/had_changed）。
  • 时间段内的统计（"过去 10 秒的最小值"、"每分钟平均"）→ 用窗口聚合 + HAVING。
  • 纯"设备 N 秒无事件"的离线告警走窗口 + 空闲检测，不必硬塞 MATCH_RECOGNIZE。
维度 模式识别（CEP） 分析函数 窗口聚合
关心什么 事件顺序 / 序列 相邻事件关系 时间段统计
状态 分区 NFA 模拟，跨多事件 跨事件状态机 窗口内聚合，触发后清空
输出 每次模式匹配产出一批 每条事件一行 窗口关闭时一批
触发 模式匹配完成时 每条事件 窗口边界
API Emit + AddSink EmitSync 或 Emit Emit + AddSink

# 一个最小例子

温度连续 3 次越过 50（防抖，避免单点抖动误报）：

SELECT * FROM stream
MATCH_RECOGNIZE (
    ORDER BY ts
    MEASURES MATCH_NUMBER() AS mn, LAST(A.temp) AS peak
    ONE ROW PER MATCH
    PATTERN (A{3})
    WITHIN '1h'
    DEFINE A AS temp > 50
)
1
2
3
4
5
6
7
8
9

输入：

{"ts": 1, "temp": 10}
{"ts": 2, "temp": 60}
{"ts": 3, "temp": 70}
{"ts": 4, "temp": 80}
{"ts": 5, "temp": 5}
1
2
3
4
5

输出（第 2–4 条构成一次 A{3} 匹配）：

{"mn": 1, "peak": 80}
1
  • PATTERN (A{3})：A 重复 3 次。
  • DEFINE A AS temp > 50：符号 A 匹配 temp > 50 的行。
  • MEASURES：匹配产出哪些列——MATCH_NUMBER() 是第几次匹配，LAST(A.temp) 是 A 的最后一次温度。
  • WITHIN '1h'：整个匹配必须落在 1 小时内（有界必需，见WITHIN）。

# 语法总览

SELECT * FROM stream
MATCH_RECOGNIZE (
    [PARTITION BY <expr> [, ...]]               -- 按字段分区，各分区独立匹配
    ORDER BY <expr> [ASC] [, ...]               -- 必需：事件排序（通常时间戳）
    [MEASURES <expr> AS <alias> [, ...]]        -- 输出列
    [ONE ROW PER MATCH | ALL ROWS PER MATCH]    -- 每次匹配产出 1 行 / 每行产出
    [AFTER MATCH SKIP ...]                      -- 匹配后从哪里继续
    PATTERN ( <模式> )                          -- 事件序列模式
    [SUBSET <name> = (<sym> [, ...]) [, ...]]   -- 符号分组
    [WITHIN '<时长>']                           -- 匹配时间窗
    DEFINE <sym> AS <cond> [, ...]              -- 各符号的匹配条件
)
1
2
3
4
5
6
7
8
9
10
11
12
子句 必需 说明
PARTITION BY 否 分区独立匹配（如每个设备各自识别）。高基数见误区 5
ORDER BY 是 事件按此字段升序处理（通常事件时间戳）
MEASURES 否 匹配产出列；省略时输出匹配末行字段
ONE / ALL ROWS PER MATCH 否 默认 ONE ROW（每次匹配 1 行）
AFTER MATCH SKIP 否 默认 PAST LAST ROW
PATTERN 是 模式表达式
SUBSET 否 把多个符号组成集合
WITHIN 否（强烈建议） 时间窗，空闲分区的部分匹配主动过期
DEFINE 否 符号条件；未定义的符号恒为真（标准）

SELECT 在 CEP 模式不参与投影

MATCH_RECOGNIZE 的输出由 MEASURES 决定，外层 SELECT 只用于写 SELECT * FROM stream MATCH_RECOGNIZE(...)。要什么列就写进 MEASURES。

# PATTERN —— 定义事件序列

写法 含义
A B C 序列：先 A 再 B 再 C
A? A 出现 0 或 1 次
A* A 出现 0 或多次（贪婪，尽量多）
A+ A 出现 1 或多次
A{3} A 恰好 3 次
A{2,} A 至少 2 次
A{2,4} A 2 到 4 次
A \| B 交替：A 或 B
(A B) 分组
PERMUTE(A, B, C) A/B/C 任意顺序各出现一次

量词后缀 ? 变懒惰（reluctant，尽量少）：A*? 选最短的 A 序列。

# 贪婪 vs 懒惰

  • A*（贪婪）：尽可能多匹配。A* B 在 A A A B 上选 A A A B（最长）。
  • A*?（懒惰）：尽可能少匹配。A*? B 选 B（最短，0 个 A 后接 B）。

何时用懒惰

默认贪婪覆盖大多数"取最长连续段"的需求。懒惰用于"尽早结束、尽快触发下一个匹配"。纯贪婪 / 纯懒惰模式完全准确；混合贪婪与懒惰量词的模式不保证逐量词优先级。

# DEFINE —— 符号的匹配条件

每个模式变量（符号）给一个布尔条件；条件可用字段、导航函数、聚合：

PATTERN (A B)
DEFINE
    A AS temp > 50,
    B AS temp < PREV(temp)      -- B 比"上一行"温度低
1
2
3
4
  • 未定义的符号恒为真（SQL 标准）——PATTERN (A B) 只给 A 定义，B 匹配任意行。
  • DEFINE 里用 PREV/NEXT/FIRST/LAST 做跨行引用（不是 LAG/LEAD，那是窗口函数）。

# MEASURES —— 输出列

MEASURES 定义每次匹配产出哪些列，可用字段、导航、聚合、两个特殊函数：

表达式 含义
A.temp 符号 A 最后一次出现的 temp（符号限定字段）
LAST(temp) / FIRST(temp) 匹配内末行 / 首行的 temp
SUM(temp) / AVG(temp) / COUNT(*) / MIN / MAX 匹配内聚合
CLASSIFIER() 当前行匹配的符号名（ALL ROWS 下每行不同）
MATCH_NUMBER() 第几次匹配（分区独立计数）
MEASURES
    MATCH_NUMBER() AS mn,
    FIRST(A.temp)  AS start_temp,
    LAST(A.temp)   AS end_temp,
    MAX(A.temp)    AS peak,
    CLASSIFIER()   AS sym
1
2
3
4
5
6

# 导航函数

函数 含义
PREV(field [, n]) 当前行往前第 n 行的字段（默认 n=1）
NEXT(field [, n]) 当前行往后第 n 行
FIRST(field [, n]) 匹配内第 n 个该字段（默认 1）
LAST(field [, n]) 匹配内倒数第 n 个

DEFINE 里"当前行"是待判定的候选行；MEASURES 里"当前行"随 ALL ROWS 推进。

# 聚合（按符号过滤）

SUM/AVG/COUNT/MIN/MAX 作用于整个匹配的行；加符号限定则只聚合该符号的行：

MEASURES SUM(A.v) AS a_total, SUM(v) AS all_total
-- 行: A.v=1, B.v=2, A.v=3 → a_total=4(只 A), all_total=6(全部)
1
2

# FINAL / RUNNING 语义

聚合与 FIRST/LAST 可加 FINAL / RUNNING 前缀：

前缀 含义
RUNNING（默认） 只算到当前行（ALL ROWS 下随推进变化）
FINAL 算整个匹配（与当前行无关）
MEASURES
    RUNNING SUM(v) AS run_total,   -- 到当前行的累计
    FINAL   SUM(v) AS match_total  -- 整个匹配的总和
1
2
3

ONE ROW PER MATCH 下只有一个末行视角，FINAL == RUNNING；ALL ROWS PER MATCH 下两者才表现出差异。

# SUBSET —— 符号分组

把多个符号组成一个集合，可在 PATTERN 和 MEASURES/DEFINE 里当整体用：

PATTERN (S)
SUBSET S = (A, B)
MEASURES SUM(S.v) AS total    -- 对 A 和 B 的行一起聚合
1
2
3

CLASSIFIER() 在 SUBSET 下返回真实成分符号（A 或 B），而非 S。支持嵌套 S2 = (S1, C)。

# WITHIN —— 时间窗

WITHIN '<时长>' 要求整个匹配落在该时长内。常见单位：'200ms' / '1s' / '5m' / '1h'。

PATTERN (A B) WITHIN '5s'     -- A 和 B 必须在 5 秒内先后到达
1
  • ORDER BY 的时间戳字段自动判单位（ns/μs/ms/s epoch）归一化。
  • 超窗的部分匹配会被主动清理：后台 sweeper 定期用 wall-clock 扫描空闲分区，清掉超窗的部分匹配——不必等下一条事件触发被动清理。

WITHIN 建议显式写

边缘内存受限，建议显式给 WITHIN 让空闲分区的部分匹配能被主动过期，避免长时间挂留。不给 WITHIN 时靠分区 LRU + 部分匹配上限兜底（见有界与内存）。

# PARTITION BY / ORDER BY

  • PARTITION BY：按字段分区，每个分区独立维护模式状态（如"每个设备各自的故障序列"）。多设备同流时必需，否则跨设备串扰。
  • ORDER BY（必需）：事件按此字段升序处理，通常是事件时间戳。
PARTITION BY deviceId ORDER BY ts
1

# ONE ROW vs ALL ROWS PER MATCH

模式 产出
ONE ROW PER MATCH（默认） 每次匹配产出 1 行（视角是匹配末行 + MEASURES）
ALL ROWS PER MATCH 每次匹配产出 N 行（匹配内每个输入行一行，CLASSIFIER()/RUNNING 随行变化）

ALL ROWS 输出量大，建议配 WITHIN 和输出节制。

# AFTER MATCH SKIP

一次匹配完成后，从哪里继续找下一次：

策略 含义
PAST LAST ROW（默认） 跳过本次匹配的最后一行之后
TO NEXT ROW 从最后一行的下一行
TO FIRST <sym> 跳到某符号的第一次出现之后
TO LAST <sym> 跳到某符号的最后一次出现之后

默认 PAST LAST ROW 保证匹配不重叠；TO NEXT ROW 允许重叠匹配。

# API —— 怎么跑起来

CEP 不能用 EmitSync

模式识别是"多进多出"的批量语义，EmitSync（同步返回单条）不支持，会报 synchronous mode does not support MATCH_RECOGNIZE。必须用 Emit（异步）+ AddSink 接收匹配产出。

package main

import (
    "fmt"
    "time"

    "github.com/rulego/streamsql"
)

func main() {
    ssql := streamsql.New()
    defer ssql.Stop()

    sql := `SELECT * FROM stream
        MATCH_RECOGNIZE (
            PARTITION BY deviceId ORDER BY ts
            MEASURES MATCH_NUMBER() AS mn, LAST(A.temp) AS peak
            ONE ROW PER MATCH
            PATTERN (A{3})
            WITHIN '1h'
            DEFINE A AS temp > 50
        )`
    if err := ssql.Execute(sql); err != nil {
        panic(err)
    }

    // CEP 用 AddSink 接收匹配（按到达顺序，建议 AddSyncSink 保序）
    ssql.AddSyncSink(func(batch []map[string]any) {
        for _, row := range batch {
            fmt.Printf("%+v\n", row)
        }
    })

    ssql.Emit(map[string]any{"deviceId": "d1", "ts": 1, "temp": 60.0})
    ssql.Emit(map[string]any{"deviceId": "d1", "ts": 2, "temp": 70.0})
    ssql.Emit(map[string]any{"deviceId": "d1", "ts": 3, "temp": 80.0})

    time.Sleep(200 * time.Millisecond) // 等异步 sink 落地
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
  • AddSyncSink：在数据处理协程内按序回调，保证匹配产出顺序；AddSink 走异步 worker pool，多 worker 不保序。
  • 流末未闭合的匹配（如未界的 A+）在 Stop 时经 Flush 冲刷产出。

# 有界与内存

边缘内存受限，模式识别内置四道闸防止病态模式 / 高频流撑爆内存：

闸门 默认 作用
WITHIN — 时间窗，主动过期超窗部分匹配
单次部分匹配行数 10000 防 A+ 单匹配无限延伸
单分区活跃部分匹配数 10000 防 A* 类状态爆炸
分区数 10000 超出按 LRU 淘汰最久未用分区

分区 LRU 淘汰后，该分区的模式状态丢失（再次出现当新分区）。PARTITION BY 用低基数键（如设备 id），别用近唯一键（时间戳、事件 id）。

# 常见误区

# 误区 1：CEP 用 EmitSync 报错

EmitSync 只支持非聚合的简单查询 / 分析函数，不支持 CEP（和窗口聚合）。CEP 用 Emit + AddSink/AddSyncSink。

# 误区 2：SELECT 写列名没效果

CEP 输出由 MEASURES 决定，外层 SELECT 不参与投影。要列就写进 MEASURES ... AS alias，外层统一 SELECT * FROM stream MATCH_RECOGNIZE(...)。

# 误区 3：DEFINE 用 LAG / LEAD

行模式导航用 PREV/NEXT/FIRST/LAST。LAG/LEAD 是窗口函数，不在 MATCH_RECOGNIZE 内。

# 误区 4：忘了 ORDER BY

ORDER BY 在 MATCH_RECOGNIZE 里必需——模式识别依赖事件顺序。省略会报错。

# 误区 5：PARTITION BY 高基数

PARTITION BY 分区数随不同键值增长。用低基数键；海量设备时靠默认 LRU 上限（10000）兜底，被淘汰的分区状态丢失（匹配重新开始）。

# 误区 6：以为 A* 选最短

A* 默认贪婪（选最长连续段）。要最短用 A*?（懒惰）。详见贪婪 vs 懒惰。

# 案例

完整的业务场景案例见 设备故障模式识别，覆盖：

  • 连续越限防抖（A{3}）
  • 升温后回落（A B，故障前兆）
  • V 型反转（A+ B+ C）
  • 开停机工作流（Start Running Stop）
  • PERMUTE 乱序事件
  • WITHIN 时间约束序列

# 性能

单核（AMD Ryzen 4800U）基准（cep 包 BenchmarkCEP_*）：

模式 吞吐 分配
序列 A B ~41 万 ops/s 20 allocs/op
星号贪婪 A* B ~27 万 ops/s 29 allocs/op
分区序列 A B + PARTITION BY ~37 万 ops/s 23 allocs/op

与分析函数 + 分区（~48 万 ops/s）同量级，适合边缘节点的 CEP 负载。

# 已知限制

  • 排除模式 {- A -}（absence / 否定）尚未实现（编译期拒绝）。
  • ALL ROWS PER MATCH WITH UNMATCHED ROWS 未实现。
  • SUBSET 作为 DEFINE 左侧（DEFINE S AS ...）未实现；右侧引用已支持。
  • 外层 ORDER BY / LIMIT 不作用于 CEP 输出。
  • 多流 CEP（跨 JOIN 多流识别）未实现。
  • 混合贪婪 / 懒惰量词的模式不保证逐量词优先级（纯贪婪 / 纯懒惰准确）。
在 GitHub 上编辑此页 (opens new window)
上次更新: 2026/07/14, 04:03:16
进阶示例
聚合函数

← 聚合函数→

Theme by Vdoing | Copyright © 2023-2026 RuleGo Team | Apache 2.0 License