摘要会写 Pattern 不等于会开发 CEP 作业——从规则到落地的完整链路是模式定义 → 检测 → 选择三阶段。这篇文章按生产视角拆解定义阶段的条件编写SimpleCondition 与 IterativeCondition 的差别后者能引用同模式已匹配事件实现连续递增这类跨事件条件、检测阶段的准备与绑定keyBy、事件时间、CEP.pattern 的 NFA 运行语义、选择阶段的三档 APIselect/flatSelect/process 怎么选、超时出口怎么写、MapString, List 怎么取数。四个完整代码案例覆盖从简单告警到迭代条件、多路输出与超时处理读完能独立搭出一个生产级 CEP 作业。关键词Flink CEP、模式定义、SimpleCondition、IterativeCondition、检测、PatternStream、NFA、选择、PatternSelectFunction、PatternFlatSelectFunction、PatternProcessFunction、TimedOutPartialMatchHandler、代码实现一、CEP 开发的完整链路不是写个 Pattern 就行CEP 篇讲了 NFA 引擎Pattern 篇讲了语法细节。但真实开发里一个 CEP 作业的代码组织是另一回事——它分三个阶段模式定义把业务规则翻译成 Pattern怎么写条件、怎么命名检测把 Pattern 挂到流上让引擎跑起来怎么准备流、怎么绑定选择从匹配结果里提取业务数据怎么选 API、怎么处理超时。三阶段各有各的 API 和坑。这篇按生产视角走一遍完整链路。二、三阶段全景阶段输入输出关键 API定义业务规则Pattern 对象begin/where/next/times/within检测Pattern 事件流PatternStreamkeyBy CEP.pattern()选择PatternStreamDataStream 侧输出select/flatSelect/process定义阶段是纯声明不执行匹配只描述规则检测阶段是引擎执行用户只看到 PatternStream选择阶段是结果落地匹配与超时两条出口都在这。三、模式定义条件编写的两种姿势3.1 SimpleCondition单事件条件最简单也最常用只根据当前事件判断PatternTransaction,TransactionpPattern.Transactionbegin(large).where(newSimpleConditionTransaction(){Overridepublicbooleanfilter(Transactiont){returnt.amount100_000;// 只看当前事件够用}});3.2 IterativeCondition迭代条件能引用已匹配事件需求一复杂就发现 SimpleCondition 不够用了——比如反洗钱的经典模式连续多笔交易金额逐笔递增判断当前事件金额比上一笔大 2 倍必须知道上一笔是多少。这时用IterativeCondition它通过ctx.getEventsForPattern(模式名)拿到同模式内之前已匹配的事件PatternTransaction,TransactionescalatingPattern.Transactionbegin(tx).where(newIterativeConditionTransaction(){Overridepublicbooleanfilter(Transactioncurrent,ContextTransactionctx){// 第一笔无条件进入if(!ctx.getEventsForPattern(tx).iterator().hasNext()){returncurrent.amount10_000;// 起点大额}// 后续事件金额必须比上一笔大 2 倍跨事件条件Transactionprevctx.getEventsForPattern(tx).iterator().next();// 同模式已匹配的上一笔returncurrent.amountprev.amount*2;}}).times(3)// 3 笔逐笔翻倍.within(Time.minutes(5));这是 CEP 表达力的分水岭SimpleCondition 看单事件IterativeCondition 看事件序列的上下文。“连续递增”“比上一笔大”首笔触发后行为变化这类规则只有 IterativeCondition 能优雅表达——用 SimpleCondition 只能靠外部状态 hack。3.3 命名规范模式名 选择阶段的取数 keybegin(start)里的名字不是装饰——它是选择阶段match.get(start)的 key。规范语义化、唯一、与选择函数里的引用完全一致。拼错一个字编译期不报错运行期 NPE。四、检测把 Pattern 挂到流上检测阶段只有三步但前两步决定一切// ① 准备keyBy 事件时间缺一不可KeyedStreamTransaction,StringkeyedtxStream.keyBy(Transaction::getAccountId)// 每账户独立检测.assignTimestampsAndWatermarks(WatermarkStrategy.TransactionforBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((t,ts)-t.getEventTs()));// ② 绑定流 模式 跳过策略 → PatternStreamPatternStreamTransactionpsCEP.pattern(keyed,escalating,AfterMatchSkipStrategy.skipPastLastEvent());// ③ 产物PatternStream 已包含匹配 超时两类记录等待选择阶段消费检测的底层就是 NFA 运行CEP 篇讲过每个 key 一个独立 NFA 实例事件按 key 路由、驱动状态转移within 定时器是状态随 checkpoint 恢复。这些对用户透明——你只看到 PatternStream。但两个准备不做检测就是错的不 keyBy跨账户的事件会串进同一个匹配不配事件时间within 超时永不触发。五、选择三档 API 与超时出口5.1 三档 API 怎么选API一条匹配输出超时写法上下文适用select1 条select(tag, timeoutFn, selectFn)无简单告警flatSelect多条flatSelect(tag, flatTimeoutFn, flatSelectFn)无告警指标process多条实现 TimedOutPartialMatchHandler完整生产主力三档都接收MapString, ListIN——按模式名取该模式匹配到的事件列表match.get(order)取下单事件match.get(pay)取支付事件times 循环时列表里有多个元素optional 模式可能缺失要判空超时部分匹配的 Map 只有已匹配的模式比如只有 “order” 没有 “pay”。5.2 完整案例一三阶段串起来的简单告警select// 需求5 分钟内连续 3 次登录失败 → 告警最简单链路// ── 定义 ──PatternLoginEvent,LoginEventpPattern.LoginEventbegin(start).where(e-e.resultFAIL).next(mid).where(e-e.resultFAIL).times(2).consecutive().within(Time.minutes(5));// ── 检测 ──PatternStreamLoginEventpsCEP.pattern(logins.keyBy(LoginEvent::getUserId),p);// ── 选择 ──DataStreamAlertalertsps.select((MapString,ListLoginEventmatch)-{LoginEventfirstmatch.get(start).get(0);// 按模式名取事件returnnewAlert(first.userId,brute-force);});5.3 完整案例二一次命中多路输出flatSelect命中一条撞库模式同时落告警和监控指标两条记录DataStreamObjectresultps.flatSelect((MapString,ListLoginEventmatch,CollectorObjectout)-{LoginEventfirstmatch.get(start).get(0);out.collect(newAlert(first.userId,brute-force));// 告警out.collect(newMetric(brute-force,1));// 指标// 一次匹配 → 两条输出下游各自消费});5.4 完整案例三process 超时处理生产主力订单超时场景匹配与超时在一个类里收口超时走侧输出// 模式下单后 10 分钟内未支付 → 超时PatternOrderEvent,OrderEventpPattern.OrderEventbegin(order).where(e-e.typeCREATE).followedBy(pay).where(e-e.typePAY).within(Time.minutes(10));PatternStreamOrderEventpsCEP.pattern(orders.keyBy(OrderEvent::getOrderId),p);OutputTagOrderEventtimeoutTagnewOutputTagOrderEvent(timeout){};DataStreamStringresultps.process(newPatternProcessFunctionOrderEvent,String(){OverridepublicvoidprocessMatch(MapString,ListOrderEventmatch,Contextctx,CollectorStringout){// 完整匹配下单 → 支付out.collect(PAID:match.get(pay).get(0).getOrderId());}OverridepublicvoidhandleTimeout(MapString,ListOrderEventpartial,longts,Contextctx)throwsException{// 超时部分匹配只有 orderpay 模式缺失// 注意这里也能侧输出——超时与匹配两条链路一个类收口ctx.output(timeoutTag,partial.get(order).get(0));}});DataStreamOrderEventtimeoutOrdersresult.getSideOutput(timeoutTag);// timeoutOrders → 关单/提醒result → 正常支付订单process 相比 select 的优势在代码组织上最明显processMatch 和 handleTimeout 是同一个类的两个方法超时逻辑就在匹配逻辑旁边不用像 select 那样把超时函数拆到另一个匿名类里。生产环境我默认选 process。5.5 迭代条件 选择的组合逐笔递增交易告警把第三节的 IterativeCondition 模式接上选择就是一个完整的反洗钱检测// 定义迭代条件见 3.2 的 escalating 模式 检测 选择DataStreamRiskAlertalertsCEP.pattern(keyedTx,escalating).flatSelect((MapString,ListTransactionmatch,CollectorRiskAlertout)-{// times(3) 循环match.get(tx) 里有 3 笔事件ListTransactiontxsmatch.get(tx);out.collect(newRiskAlert(txs.get(0).accountId,escalating,txs));});六、实战避坑清单迭代条件别滥用IterativeCondition 每次评估都要遍历已匹配事件大循环模式 高频事件会放大开销——只在确实需要跨事件条件时用match.get() 前先想 optionaloptional 模式可能缺失直接 get().get(0) 会 NPE超时 Map 缺模式handleTimeout 里的 Map 只有已匹配的模式别拿还没到的模式取数模式名一致定义与选择两处引用必须完全一致建议抽 static final 常量检测前准备两行不能省keyBy assignTimestampsAndWatermarks漏一个语义就错选择函数别做重活每条匹配调用一次里面查库/调用外部系统会拖慢整个 PatternStreamprocess 优先新代码默认 PatternProcessFunction模板代码和 select 差不多但超时与上下文能力完整。七、总结我的判断CEP 开发的完整链路可以用一句话概括定义是规则、检测是引擎、选择是出口。三个阶段的关注点完全不同——定义阶段想清楚条件怎么写Simple vs Iterative检测阶段做好两个准备keyBy 事件时间选择阶段定好出口怎么落select/flatSelect/process 超时侧输出。三条实操建议新作业默认 processPatternProcessFunction 一个类收口匹配与超时后续加侧输出/时间戳不用改结构跨事件条件用 IterativeCondition 而不是外部状态 hack逐笔递增、比上一笔大这类规则迭代条件是原生姿势外部状态要自己管生命周期checkpoint 还容易漏超时出口必须配handleTimeout 或 timeoutFn 不写超时的部分匹配静默丢弃——对下单未支付这类业务就是数据丢失。
Flink高级之模式定义、检测与选择代码实现：CEP开发的完整三阶段
摘要会写 Pattern 不等于会开发 CEP 作业——从规则到落地的完整链路是模式定义 → 检测 → 选择三阶段。这篇文章按生产视角拆解定义阶段的条件编写SimpleCondition 与 IterativeCondition 的差别后者能引用同模式已匹配事件实现连续递增这类跨事件条件、检测阶段的准备与绑定keyBy、事件时间、CEP.pattern 的 NFA 运行语义、选择阶段的三档 APIselect/flatSelect/process 怎么选、超时出口怎么写、MapString, List 怎么取数。四个完整代码案例覆盖从简单告警到迭代条件、多路输出与超时处理读完能独立搭出一个生产级 CEP 作业。关键词Flink CEP、模式定义、SimpleCondition、IterativeCondition、检测、PatternStream、NFA、选择、PatternSelectFunction、PatternFlatSelectFunction、PatternProcessFunction、TimedOutPartialMatchHandler、代码实现一、CEP 开发的完整链路不是写个 Pattern 就行CEP 篇讲了 NFA 引擎Pattern 篇讲了语法细节。但真实开发里一个 CEP 作业的代码组织是另一回事——它分三个阶段模式定义把业务规则翻译成 Pattern怎么写条件、怎么命名检测把 Pattern 挂到流上让引擎跑起来怎么准备流、怎么绑定选择从匹配结果里提取业务数据怎么选 API、怎么处理超时。三阶段各有各的 API 和坑。这篇按生产视角走一遍完整链路。二、三阶段全景阶段输入输出关键 API定义业务规则Pattern 对象begin/where/next/times/within检测Pattern 事件流PatternStreamkeyBy CEP.pattern()选择PatternStreamDataStream 侧输出select/flatSelect/process定义阶段是纯声明不执行匹配只描述规则检测阶段是引擎执行用户只看到 PatternStream选择阶段是结果落地匹配与超时两条出口都在这。三、模式定义条件编写的两种姿势3.1 SimpleCondition单事件条件最简单也最常用只根据当前事件判断PatternTransaction,TransactionpPattern.Transactionbegin(large).where(newSimpleConditionTransaction(){Overridepublicbooleanfilter(Transactiont){returnt.amount100_000;// 只看当前事件够用}});3.2 IterativeCondition迭代条件能引用已匹配事件需求一复杂就发现 SimpleCondition 不够用了——比如反洗钱的经典模式连续多笔交易金额逐笔递增判断当前事件金额比上一笔大 2 倍必须知道上一笔是多少。这时用IterativeCondition它通过ctx.getEventsForPattern(模式名)拿到同模式内之前已匹配的事件PatternTransaction,TransactionescalatingPattern.Transactionbegin(tx).where(newIterativeConditionTransaction(){Overridepublicbooleanfilter(Transactioncurrent,ContextTransactionctx){// 第一笔无条件进入if(!ctx.getEventsForPattern(tx).iterator().hasNext()){returncurrent.amount10_000;// 起点大额}// 后续事件金额必须比上一笔大 2 倍跨事件条件Transactionprevctx.getEventsForPattern(tx).iterator().next();// 同模式已匹配的上一笔returncurrent.amountprev.amount*2;}}).times(3)// 3 笔逐笔翻倍.within(Time.minutes(5));这是 CEP 表达力的分水岭SimpleCondition 看单事件IterativeCondition 看事件序列的上下文。“连续递增”“比上一笔大”首笔触发后行为变化这类规则只有 IterativeCondition 能优雅表达——用 SimpleCondition 只能靠外部状态 hack。3.3 命名规范模式名 选择阶段的取数 keybegin(start)里的名字不是装饰——它是选择阶段match.get(start)的 key。规范语义化、唯一、与选择函数里的引用完全一致。拼错一个字编译期不报错运行期 NPE。四、检测把 Pattern 挂到流上检测阶段只有三步但前两步决定一切// ① 准备keyBy 事件时间缺一不可KeyedStreamTransaction,StringkeyedtxStream.keyBy(Transaction::getAccountId)// 每账户独立检测.assignTimestampsAndWatermarks(WatermarkStrategy.TransactionforBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((t,ts)-t.getEventTs()));// ② 绑定流 模式 跳过策略 → PatternStreamPatternStreamTransactionpsCEP.pattern(keyed,escalating,AfterMatchSkipStrategy.skipPastLastEvent());// ③ 产物PatternStream 已包含匹配 超时两类记录等待选择阶段消费检测的底层就是 NFA 运行CEP 篇讲过每个 key 一个独立 NFA 实例事件按 key 路由、驱动状态转移within 定时器是状态随 checkpoint 恢复。这些对用户透明——你只看到 PatternStream。但两个准备不做检测就是错的不 keyBy跨账户的事件会串进同一个匹配不配事件时间within 超时永不触发。五、选择三档 API 与超时出口5.1 三档 API 怎么选API一条匹配输出超时写法上下文适用select1 条select(tag, timeoutFn, selectFn)无简单告警flatSelect多条flatSelect(tag, flatTimeoutFn, flatSelectFn)无告警指标process多条实现 TimedOutPartialMatchHandler完整生产主力三档都接收MapString, ListIN——按模式名取该模式匹配到的事件列表match.get(order)取下单事件match.get(pay)取支付事件times 循环时列表里有多个元素optional 模式可能缺失要判空超时部分匹配的 Map 只有已匹配的模式比如只有 “order” 没有 “pay”。5.2 完整案例一三阶段串起来的简单告警select// 需求5 分钟内连续 3 次登录失败 → 告警最简单链路// ── 定义 ──PatternLoginEvent,LoginEventpPattern.LoginEventbegin(start).where(e-e.resultFAIL).next(mid).where(e-e.resultFAIL).times(2).consecutive().within(Time.minutes(5));// ── 检测 ──PatternStreamLoginEventpsCEP.pattern(logins.keyBy(LoginEvent::getUserId),p);// ── 选择 ──DataStreamAlertalertsps.select((MapString,ListLoginEventmatch)-{LoginEventfirstmatch.get(start).get(0);// 按模式名取事件returnnewAlert(first.userId,brute-force);});5.3 完整案例二一次命中多路输出flatSelect命中一条撞库模式同时落告警和监控指标两条记录DataStreamObjectresultps.flatSelect((MapString,ListLoginEventmatch,CollectorObjectout)-{LoginEventfirstmatch.get(start).get(0);out.collect(newAlert(first.userId,brute-force));// 告警out.collect(newMetric(brute-force,1));// 指标// 一次匹配 → 两条输出下游各自消费});5.4 完整案例三process 超时处理生产主力订单超时场景匹配与超时在一个类里收口超时走侧输出// 模式下单后 10 分钟内未支付 → 超时PatternOrderEvent,OrderEventpPattern.OrderEventbegin(order).where(e-e.typeCREATE).followedBy(pay).where(e-e.typePAY).within(Time.minutes(10));PatternStreamOrderEventpsCEP.pattern(orders.keyBy(OrderEvent::getOrderId),p);OutputTagOrderEventtimeoutTagnewOutputTagOrderEvent(timeout){};DataStreamStringresultps.process(newPatternProcessFunctionOrderEvent,String(){OverridepublicvoidprocessMatch(MapString,ListOrderEventmatch,Contextctx,CollectorStringout){// 完整匹配下单 → 支付out.collect(PAID:match.get(pay).get(0).getOrderId());}OverridepublicvoidhandleTimeout(MapString,ListOrderEventpartial,longts,Contextctx)throwsException{// 超时部分匹配只有 orderpay 模式缺失// 注意这里也能侧输出——超时与匹配两条链路一个类收口ctx.output(timeoutTag,partial.get(order).get(0));}});DataStreamOrderEventtimeoutOrdersresult.getSideOutput(timeoutTag);// timeoutOrders → 关单/提醒result → 正常支付订单process 相比 select 的优势在代码组织上最明显processMatch 和 handleTimeout 是同一个类的两个方法超时逻辑就在匹配逻辑旁边不用像 select 那样把超时函数拆到另一个匿名类里。生产环境我默认选 process。5.5 迭代条件 选择的组合逐笔递增交易告警把第三节的 IterativeCondition 模式接上选择就是一个完整的反洗钱检测// 定义迭代条件见 3.2 的 escalating 模式 检测 选择DataStreamRiskAlertalertsCEP.pattern(keyedTx,escalating).flatSelect((MapString,ListTransactionmatch,CollectorRiskAlertout)-{// times(3) 循环match.get(tx) 里有 3 笔事件ListTransactiontxsmatch.get(tx);out.collect(newRiskAlert(txs.get(0).accountId,escalating,txs));});六、实战避坑清单迭代条件别滥用IterativeCondition 每次评估都要遍历已匹配事件大循环模式 高频事件会放大开销——只在确实需要跨事件条件时用match.get() 前先想 optionaloptional 模式可能缺失直接 get().get(0) 会 NPE超时 Map 缺模式handleTimeout 里的 Map 只有已匹配的模式别拿还没到的模式取数模式名一致定义与选择两处引用必须完全一致建议抽 static final 常量检测前准备两行不能省keyBy assignTimestampsAndWatermarks漏一个语义就错选择函数别做重活每条匹配调用一次里面查库/调用外部系统会拖慢整个 PatternStreamprocess 优先新代码默认 PatternProcessFunction模板代码和 select 差不多但超时与上下文能力完整。七、总结我的判断CEP 开发的完整链路可以用一句话概括定义是规则、检测是引擎、选择是出口。三个阶段的关注点完全不同——定义阶段想清楚条件怎么写Simple vs Iterative检测阶段做好两个准备keyBy 事件时间选择阶段定好出口怎么落select/flatSelect/process 超时侧输出。三条实操建议新作业默认 processPatternProcessFunction 一个类收口匹配与超时后续加侧输出/时间戳不用改结构跨事件条件用 IterativeCondition 而不是外部状态 hack逐笔递增、比上一笔大这类规则迭代条件是原生姿势外部状态要自己管生命周期checkpoint 还容易漏超时出口必须配handleTimeout 或 timeoutFn 不写超时的部分匹配静默丢弃——对下单未支付这类业务就是数据丢失。