📚 课程目录第4章 Flink CEP核心API

4. Flink CEP 核心API：Pattern API详解

Pattern API 是 Flink CEP 的“规则语言”，它通过链式调用的方式定义事件匹配的时序逻辑、量词约束与时间窗口。在异常交易样本构建中，Pattern API 决定了“什么样的交易序列会被判定为可疑”。本章将深入解析 begin、next、followedBy、within 等核心操作符，并结合金融风控场景给出可复用的模式模板。

4.1 模式定义三要素：开始、连接与时间

一个完整的 CEP 模式由三部分构成：起始模式（begin）、后继连接（next / followedBy / followedByAny）和时间约束（within）。下表总结了各操作符的语义与适用场景：

操作符	语义	匹配逻辑	典型风控场景
begin	定义模式起点	从第一个满足条件的事件开始匹配	“首次大额转账”作为异常序列的触发事件
next	严格连续	两个事件必须紧邻，中间不能有其他事件	“登录失败”后立即“修改密码”的暴力破解行为
followedBy	宽松连续	允许中间跳过任意非匹配事件	“小额试探”后任意时刻出现“大额转出”的洗钱模式
followedByAny	非确定性宽松连续	允许重叠匹配，同一事件可参与多个匹配序列	多账户关联的“分散转入-集中转出”网状交易
within	时间窗口	整个匹配序列必须在指定时间内完成	“30分钟内连续3笔失败交易”的撞库检测

实战建议：在异常交易样本构建中，followedBy 的使用频率远高于 next。因为真实交易流中往往夹杂大量正常事件，严格连续会导致漏报。仅在需要检测“紧邻操作”时（如连续两次密码错误）才使用 next。

4.2 起始模式：begin

begin 是模式链的入口，它定义了一个“种子事件”。所有后续操作符都基于这个种子事件进行扩展。begin 支持传入事件类型或条件谓词：

// 方式一：按事件类型匹配
Pattern<TransactionEvent, ?> start = Pattern.<TransactionEvent>begin("first")
    .where(new SimpleCondition<TransactionEvent>() {
        @Override
        public boolean filter(TransactionEvent event) {
            return event.getAmount() > 50000;
        }
    });

// 方式二：组合条件（金额 > 5万 且 跨行交易）
Pattern<TransactionEvent, ?> start = Pattern.<TransactionEvent>begin("first")
    .where(event -> event.getAmount() > 50000)
    .where(event -> "CROSS_BANK".equals(event.getChannel()));

在风控规则中，begin 通常对应“触发条件”——例如单笔金额超过阈值、交易对手为黑名单账户、或设备指纹异常。一个良好的 begin 条件能大幅降低后续匹配的计算压力。

4.3 严格连续与宽松连续：next vs followedBy

这是 Pattern API 中最容易混淆的一对操作符。它们的核心区别在于“是否允许中间事件存在”。

next：A.next(B) 表示 A 事件发生后，下一个事件必须是 B。如果中间出现其他事件，匹配失败。
followedBy：A.followedBy(B) 表示 A 事件发生后，后续任意位置出现 B 即可，中间可以插入任意数量的事件。

以下示例展示了两种模式在相同事件流上的不同匹配结果：

// 事件流（按时间顺序）：
// T1: 登录成功   T2: 查询余额   T3: 大额转账   T4: 修改密码

// 模式1：next 严格连续
Pattern.<TransactionEvent>begin("login")
    .where(event -> "LOGIN".equals(event.getType()))
    .next("transfer")
    .where(event -> event.getAmount() > 100000);
// 结果：不匹配（T1和T3之间插入了T2）

// 模式2：followedBy 宽松连续
Pattern.<TransactionEvent>begin("login")
    .where(event -> "LOGIN".equals(event.getType()))
    .followedBy("transfer")
    .where(event -> event.getAmount() > 100000);
// 结果：匹配（T1 → T3，跳过T2）

性能警告：followedBy 由于需要维护中间状态的“跳过”能力，其状态规模随模式长度呈指数增长。在长模式（超过5个节点）中，优先考虑 next 或使用 consecutive() 修饰符来限制跳过行为。

4.4 时间窗口约束：within

within 为整个模式序列施加一个时间上限。如果从第一个匹配事件开始到最后一个匹配事件结束的时间差超过指定值，该序列被丢弃。这是防止“跨日/跨周”的旧事件被错误关联的关键手段。

// 检测“30分钟内，同一账户先小额试探，再大额转出”
Pattern.<TransactionEvent>begin("probe")
    .where(event -> event.getAmount() < 100)
    .followedBy("largeOut")
    .where(event -> event.getAmount() > 50000)
    .within(Time.minutes(30));

within 的时间单位支持 Time.seconds()、Time.minutes()、Time.hours() 和 Time.days()。在异常交易样本中，建议将窗口控制在 5 分钟到 2 小时之间，过长的时间窗口会引入大量噪声。

4.5 模式连接流程总览

下图展示了从 begin 到 within 的完整模式构建流程，以及各操作符在事件流上的作用位置：

4.6 组合模式：多分支与循环匹配

在复杂异常交易检测中，单一线性模式往往不够。Pattern API 支持通过 or()、until() 和 times() 构建多分支与循环逻辑：

// 检测“同一设备在5分钟内，使用3个不同账户登录失败”
Pattern.<LoginEvent>begin("first")
    .where(event -> "FAIL".equals(event.getResult()))
    .followedBy("second")
    .where(event -> "FAIL".equals(event.getResult()))
    .where(event -> !event.getAccount().equals(
        ((LoginEvent)getEvents().get("first").get(0)).getAccount()))
    .followedBy("third")
    .where(event -> "FAIL".equals(event.getResult()))
    .where(event -> !event.getAccount().equals(
        ((LoginEvent)getEvents().get("first").get(0)).getAccount()))
    .where(event -> !event.getAccount().equals(
        ((LoginEvent)getEvents().get("second").get(0)).getAccount()))
    .within(Time.minutes(5));

上述模式使用了 getEvents() 方法引用前序事件的值，实现了“不同账户”的约束。这是 CEP 模式中实现跨事件字段比较的标准做法。

4.7 操作符选择决策树

在实际构建异常交易样本时，可以按照以下决策树快速选择正确的操作符：

4.8 常见陷阱与避坑指南

陷阱1：在 followedBy 后忘记加 within，导致跨天事件被匹配。解决方案：始终为模式设置合理的时间窗口。
陷阱2：在 next 模式中插入 where 条件时，误将中间事件过滤掉导致匹配失败。解决方案：使用 notNext() 或 notFollowedBy() 显式排除。
陷阱3：多个 where 条件之间是 AND 关系，不是 OR。需要 OR 逻辑时使用 or() 方法。
陷阱4：begin 中使用了 times(2) 会导致模式起始点变为“两个连续事件”，而非单个事件。解决方案：将重复次数放在后续节点上。

  样本构建黄金法则：在编写 Pattern 时，先用伪代码写出“如果……然后……在……时间内”的自然语言规则，再逐句翻译为 Pattern API 调用链。这能有效减少逻辑遗漏。