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

4. Flink CEP 核心API：Pattern API详解、select与flatSelect、PatternStream

在风控巡检场景中，Flink CEP 的核心能力体现在对事件流的模式匹配与告警生成。本章将深入解析 Pattern API 的构造语法、select / flatSelect 的输出差异，以及 PatternStream 的运行时结构。掌握这些 API 是编写高效风控规则的基础。

4.1 Pattern API 构造语法

Pattern API 采用链式调用，从定义第一个模式开始，逐步添加后续模式与约束条件。其核心分为三部分：模式定义、条件约束、连续性策略。

API 方法	用途	风控示例
`Pattern.begin("name")`	定义模式序列的起始节点	`Pattern.begin("login")`
`.where(condition)`	添加过滤条件（简单条件）	`.where(_.amount > 10000)`
`.or(condition)`	添加或逻辑条件	`.or(_.ip != "internal")`
`.until(condition)`	循环模式的终止条件	`.until(_.status == "verified")`
`.times(n)`	指定事件出现次数	`.times(3)` 表示连续3次
`.oneOrMore()`	一次或多次匹配	用于高频交易检测
`.greedy()`	贪婪匹配模式	尽可能多地匹配事件
`.next("name")`	严格连续（必须紧邻）	连续失败登录
`.followedBy("name")`	宽松连续（允许中间有事件）	登录后任意时刻转账
`.followedByAny("name")`	非确定性宽松连续	同一设备多次操作
`.notNext()`	否定模式（严格）	登录后没有验证
`.notFollowedBy()`	否定模式（宽松）	转账前没有风控检查
`.within(time)`	时间窗口约束	`.within(Time.minutes(5))`

风控实践提示： 在异常交易检测中，.followedBy 比 .next 更实用，因为真实业务事件之间往往存在其他无关事件（如心跳、查询）。使用 .within 限制时间窗口可以避免跨天匹配导致的误报。

4.2 模式序列与连续性策略

连续性策略决定了事件之间如何连接。风控场景中，错误的选择会导致漏报或误报。以下对比三种核心策略：

策略	语义	匹配要求	风控适用场景
严格连续 (`.next`)	事件必须紧邻出现	中间不能有任何事件	连续失败登录（无间隔）
宽松连续 (`.followedBy`)	允许中间有任意事件	后续事件在之前事件之后即可	登录→转账（中间可能有查询）
非确定性宽松连续 (`.followedByAny`)	允许跳过事件并重复匹配	同一事件可参与多个匹配	同一设备多次操作触发多条规则

避坑： 使用 .followedByAny 时，匹配结果数量会指数级增长，容易导致状态膨胀。在风控高吞吐场景下，建议优先使用 .followedBy，仅在需要多路匹配时谨慎使用 .followedByAny。

4.3 PatternStream 运行时结构

当 Pattern 定义完成后，通过 CEP.pattern(inputStream, pattern) 方法生成 PatternStream。PatternStream 并不是一个独立的数据流，而是 CEP 内部状态机与匹配缓冲区的逻辑封装。

PatternStream 内部维护一个 NFA（非确定性有限自动机），每个输入事件都会触发状态转移。匹配缓冲区保存所有部分匹配的序列，当到达接收状态时，将完整匹配输出到下游。

4.4 select 与 flatSelect 详解

PatternStream 提供了两种输出算子：select 和 flatSelect。两者的核心区别在于 输出粒度 和 处理能力。

4.4.1 select：单条匹配输出

select 接收一个 PatternSelectFunction，每个匹配的序列只输出一条结果。适用于每个模式匹配只需生成一条告警的场景。

// 风控示例：检测单笔大额交易
Pattern<Transaction, Transaction> pattern = Pattern
    .<Transaction>begin("large_trade")
    .where(tx -> tx.getAmount() > 50000)
    .within(Time.minutes(1));

PatternStream<Transaction> patternStream = CEP.pattern(transactions, pattern);

DataStream<Alert> alerts = patternStream.select(
    (Map<String, List<Transaction>> patternMap) -> {
        Transaction tx = patternMap.get("large_trade").get(0);
        return new Alert(tx.getId(), "大额交易告警", tx.getAmount());
    }
);

4.4.2 flatSelect：多条结果输出

flatSelect 接收 PatternFlatSelectFunction，允许一个匹配序列输出 零条、一条或多条 结果。适用于需要根据匹配内容动态生成多条告警或执行分支逻辑的场景。

// 风控示例：高频交易检测，按不同阈值输出多条告警
DataStream<Alert> alerts = patternStream.flatSelect(
    (Map<String, List<Transaction>> patternMap, Collector<Alert> out) -> {
        List<Transaction> trades = patternMap.get("frequent_trades");
        double totalAmount = trades.stream().mapToDouble(Transaction::getAmount).sum();
        int count = trades.size();

        // 根据交易次数和金额输出不同级别的告警
        if (count >= 5) {
            out.collect(new Alert("HIGH", "高频交易次数超标", count));
        }
        if (totalAmount > 100000) {
            out.collect(new Alert("CRITICAL", "高频交易金额超标", totalAmount));
        }
        // 如果都不满足，则不输出任何告警
    }
);

4.5 select 与 flatSelect 对比

对比维度	select	flatSelect
输出数量	每个匹配严格输出 1 条	每个匹配可输出 0 ~ N 条
函数接口	`PatternSelectFunction`	`PatternFlatSelectFunction`
Collector 参数	无，直接返回结果	有，通过 `out.collect()` 输出
适用场景	一对一映射（如单条告警）	一对多映射（如分级告警、分支处理）
风控典型用法	命中规则直接生成告警	根据命中详情动态调整告警级别
性能开销	较低	略高（需创建 Collector 实例）

  选型建议： 在风控巡检中，如果规则逻辑简单（如“金额超过阈值则告警”），优先使用 select 以降低开销。如果规则需要根据匹配详情做分级、聚合或条件分支（如“连续3次失败后，根据IP归属地决定告警级别”），则使用 flatSelect 更灵活。

4.6 侧输出（Side Output）在风控中的应用

除了主输出流，PatternStream 还支持 侧输出（Side Output），用于将不同类型的匹配结果分流。这在风控中非常实用，例如将“高风险告警”和“低风险告警”分别输出到不同的下游处理逻辑。

// 定义侧输出标签
OutputTag<Alert> highRiskTag = new OutputTag<Alert>("high-risk") {};
OutputTag<Alert> lowRiskTag = new OutputTag<Alert>("low-risk") {};

// 使用 flatSelect 结合侧输出
DataStream<Alert> mainStream = patternStream.flatSelect(
    (Map<String, List<Transaction>> patternMap, Collector<Alert> out) -> {
        // 主输出：默认告警
        out.collect(new Alert("DEFAULT", "交易异常", 0));
    },
    // 侧输出：高风险
    highRiskTag,
    (Map<String, List<Transaction>> patternMap, Collector<Alert> out) -> {
        out.collect(new Alert("HIGH", "高风险交易", 0));
    },
    // 侧输出：低风险
    lowRiskTag,
    (Map<String, List<Transaction>> patternMap, Collector<Alert> out) -> {
        out.collect(new Alert("LOW", "低风险交易", 0));
    }
);

// 获取侧输出流
DataStream<Alert> highRiskStream = mainStream.getSideOutput(highRiskTag);
DataStream<Alert> lowRiskStream = mainStream.getSideOutput(lowRiskTag);

4.7 本章小结

Pattern API 通过链式调用定义模式序列，核心方法包括 begin、where、next、followedBy、times、within 等。
连续性策略 选择直接影响匹配结果：严格连续用于紧邻事件，宽松连续用于允许间隔，非确定性宽松连续用于多路匹配。
PatternStream 是 NFA 状态机与匹配缓冲区的封装，运行时包含状态转移、部分匹配缓存和输出策略。
select 输出单条结果，适合简单告警；flatSelect 输出多条结果，适合分级告警和分支逻辑。
侧输出 支持将不同风险等级的告警分流，提升风控系统的处理灵活性。