📚 课程目录第2章 CEP模式API入门

2. CEP模式API入门：Pattern API核心概念、个体模式与组合模式

交易风控规则的本质是“在连续事件流中识别特定序列”。Flink CEP 的 Pattern API 正是用来描述这种序列的声明式编程接口。本章将系统讲解 Pattern API 的三大支柱：核心概念、个体模式（begin/where）与组合模式（next/followedBy），并通过风控场景的代码示例帮助你快速上手。

2.1 Pattern API 核心概念

Pattern API 将规则定义抽象为“模式图”（Pattern Graph），每个模式节点代表一个事件条件，节点间的边代表事件间的时序关系。理解以下三个核心概念是掌握 CEP 的基础：

概念	说明	风控类比
模式（Pattern）	描述一个事件应满足的条件，可包含量词（次数、贪婪等）	“单笔交易金额 > 5000”是一个模式
模式序列（Pattern Sequence）	多个模式按时间或逻辑顺序组合成的链	“先登录失败3次，再转账成功”是一个序列
模式匹配（Pattern Match）	事件流中满足整个模式序列的一组事件	检测到一次完整的“撞库+盗转”攻击

关键认知：Pattern API 不直接处理事件流，而是先构建一个“规则模板”，再由 CEP 引擎将模板应用到 DataStream 上执行匹配。规则与数据分离，是风控规则可配置化的基础。

2.2 个体模式：begin 与 where

个体模式（Individual Pattern）是模式序列的起点和基本单元。通过 begin() 定义第一个模式，where() 设置过滤条件。

2.2.1 begin() — 模式序列的起点

每个模式序列必须以 begin() 开头，参数为模式名称，用于后续结果提取。

// 定义第一个模式：检测单笔交易金额超过 10000
Pattern<Transaction, Transaction> start = Pattern
    .<Transaction>begin("large_transaction")
    .where(new SimpleCondition<Transaction>() {
        @Override
        public boolean filter(Transaction event) {
            return event.getAmount() > 10000;
        }
    });

2.2.2 where() — 条件过滤

where() 支持多种条件定义方式，适用于不同风控场景：

条件类型	API 方法	风控示例
简单条件	`where(SimpleCondition)`	交易金额 > 阈值
组合条件	`where(Condition).or(Condition)`	金额 > 10000 或跨行交易
迭代条件	`where(IterativeCondition)`	当前事件与之前匹配事件比较
子类型条件	`subtype(Class)`	只处理特定类型的事件子类

  风控最佳实践：在 where() 中优先使用 IterativeCondition，因为它可以访问之前已匹配的事件上下文，适合实现“同一设备短时间内多次尝试”等跨事件规则。

// 迭代条件示例：检测同一设备在 5 分钟内第 2 次交易
Pattern<Transaction, Transaction> deviceRepeat = Pattern
    .<Transaction>begin("first_tx")
    .where(new IterativeCondition<Transaction>() {
        @Override
        public boolean filter(Transaction current, Context<Transaction> ctx) {
            // 获取之前匹配的事件
            for (Transaction prev : ctx.getEventsForPattern("first_tx")) {
                if (prev.getDeviceId().equals(current.getDeviceId())) {
                    return true; // 同一设备重复交易
                }
            }
            return false;
        }
    });

2.3 组合模式：next 与 followedBy

个体模式只能描述单一事件条件，风控规则通常需要多个事件的时序组合。Pattern API 提供两种核心组合方式：严格连续（next） 和 宽松连续（followedBy）。

组合操作	语义	中间允许无关事件	风控典型场景
`next()`	严格连续：模式 B 必须紧跟在模式 A 之后	不允许	“登录成功 → 立即转账”的紧邻操作
`followedBy()`	宽松连续：模式 B 在模式 A 之后出现即可	允许	“登录失败 → 后续任意时刻转账成功”
`followedByAny()`	非确定性宽松连续：允许跳过事件，且可重复匹配	允许	“多次尝试登录 → 任意一次成功”

常见误区：next() 要求事件在时间上严格相邻，如果中间插入了其他事件（即使不满足条件），匹配也会失败。在风控场景中，followedBy() 更常用，因为攻击行为之间往往夹杂正常事件。

2.4 实战：构建一个简单的风控模式序列

下面结合个体模式和组合模式，构建一个“异常登录后大额转账”的检测规则：

// 步骤1：定义事件类型
public class LoginEvent {
    public String userId;
    public boolean success;
    public String ip;
    // getter/setter 省略
}

public class TransferEvent {
    public String userId;
    public double amount;
    public String targetAccount;
    // getter/setter 省略
}

// 步骤2：构建模式序列
Pattern<Object, ?> fraudPattern = Pattern
    .<Object>begin("failed_login")
    .subtype(LoginEvent.class)
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent login) {
            return !login.isSuccess();  // 登录失败
        }
    })
    .followedBy("large_transfer")
    .subtype(TransferEvent.class)
    .where(new SimpleCondition<TransferEvent>() {
        @Override
        public boolean filter(TransferEvent transfer) {
            return transfer.getAmount() > 50000;  // 大额转账
        }
    })
    .within(Time.minutes(10));  // 10 分钟内发生

// 步骤3：应用到 DataStream
DataStream<Object> inputStream = ...;  // 合并后的登录+交易流
PatternStream<Object> patternStream = CEP.pattern(inputStream, fraudPattern);

// 步骤4：输出告警
patternStream.select(new PatternSelectFunction<Object, Alert>() {
    @Override
    public Alert select(Map<String, List<Object>> pattern) throws Exception {
        LoginEvent failedLogin = (LoginEvent) pattern.get("failed_login").get(0);
        TransferEvent largeTransfer = (TransferEvent) pattern.get("large_transfer").get(0);
        return new Alert(
            "高风险操作：登录失败后大额转账",
            failedLogin.getUserId(),
            largeTransfer.getAmount()
        );
    }
}).print();

代码要点：

使用 subtype() 区分不同类型的事件，避免类型转换异常
followedBy() 比 next() 更适合风控场景，因为攻击行为之间可能夹杂正常事件
within() 设置时间窗口，防止历史事件无限期参与匹配

2.5 本章小结

Pattern API 核心概念：模式、模式序列、模式匹配，三者构成 CEP 规则的基础框架
个体模式：begin() 定义起点，where() 设置条件，IterativeCondition 支持跨事件上下文
组合模式：next() 严格连续，followedBy() 宽松连续，followedByAny() 非确定性宽松连续
实战要点：合理选择组合方式、设置时间窗口、使用 subtype() 处理多事件类型