📚 课程目录第4章 Flink CEP API详解

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

Flink CEP 的核心在于将业务规则转化为可执行的匹配逻辑。本章深入剖析 CEP 的三大核心 API 层：Pattern API（规则定义）、PatternStream（匹配流生成）以及 select / flatSelect（结果提取）。掌握这三层是构建实时风控人群切片的基础。

4.1 Pattern API：定义匹配规则

Pattern API 是 CEP 的“规则语言”，采用链式调用构建复杂事件序列。每个 Pattern 由 个体模式 和 组合条件 构成。

4.1.1 个体模式（Individual Pattern）

个体模式定义单个事件的匹配条件，分为 单例模式（匹配一次）和 循环模式（匹配多次）。

方法	说明	示例
`.where(condition)`	定义事件过滤条件，支持 `SimpleCondition` 或 `IterativeCondition`	`.where(event -> event.getAmount() > 1000)`
`.or(condition)`	添加“或”逻辑条件	`.where(...).or(...)`
`.times(n)`	指定事件出现次数（精确匹配 n 次）	`.times(3)`
`.times(n, m)`	指定事件出现次数范围（n 到 m 次）	`.times(2, 5)`
`.oneOrMore()`	匹配一次或多次（贪婪）	`.oneOrMore()`
`.optional()`	标记该模式为可选	`.times(2).optional()`
`.greedy()`	启用贪婪匹配（尽可能多地匹配事件）	`.oneOrMore().greedy()`

风控实践提示： 在“短时间高频交易”场景中，使用 .oneOrMore().greedy() 配合时间约束，可以捕获连续发生的异常交易序列，避免漏掉中间事件。

4.1.2 组合模式与连续性策略

多个个体模式通过 连续性策略 连接，决定事件之间的匹配关系。

策略	方法	含义	风控场景
严格连续	`.next()`	两个事件必须紧挨着出现，中间不能有其他事件	“登录后立即支付”的强关联检测
宽松连续	`.followedBy()`	允许中间出现其他事件，但顺序必须保持	“浏览商品 → 加入购物车 → 支付”的漏斗分析
非确定性宽松连续	`.followedByAny()`	允许中间出现其他事件，且允许跳过匹配事件	“同一设备多次登录不同账号”的关联检测
否定模式	`.notNext()` / `.notFollowedBy()`	指定某个事件在特定位置不能出现	“支付前没有验证手机”的异常检测

  关键区别： .next() 要求事件在时间线上严格相邻；.followedBy() 只要求顺序正确，不要求相邻；.followedByAny() 则允许跳过已匹配的事件，常用于“一对多”的关联分析。

4.1.3 时间约束与窗口

在风控场景中，事件序列必须在特定时间窗口内完成，否则视为无效。

// 定义规则：5分钟内，同一用户连续失败登录3次以上
Pattern<LoginEvent, LoginEvent> failPattern = Pattern
    .<LoginEvent>begin("first")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return event.getStatus().equals("FAIL");
        }
    })
    .times(3)
    .consecutive()  // 要求严格连续失败
    .within(Time.minutes(5));  // 整体时间窗口

注意： .within() 作用于整个模式序列，而非单个事件。如果序列总时长超过窗口，整个匹配会被丢弃。在风控中，窗口大小直接影响召回率和误报率，需要根据业务场景精细调优。

4.2 PatternStream：从规则到匹配流

将定义好的 Pattern 应用到 DataStream 上，生成 PatternStream。PatternStream 是 CEP 的中间产物，代表“可能匹配的事件序列流”。

// 将规则应用到事件流上
DataStream<LoginEvent> loginStream = ...;
Pattern<LoginEvent, LoginEvent> pattern = ...;

PatternStream<LoginEvent> patternStream = CEP.pattern(
    loginStream,
    pattern
);

PatternStream 内部维护了 NFA（非确定性有限自动机） 状态机，每个事件进入后，NFA 会尝试进行状态转移。匹配成功的事件序列会被暂存，等待后续提取。

4.3 select 与 flatSelect：提取匹配结果

PatternStream 提供了两种提取匹配结果的方式：select（一对一输出）和 flatSelect（一对多输出）。

4.3.1 select：标准匹配提取

select() 接收一个 PatternSelectFunction，每个匹配序列输出一条结果。适用于“一个匹配对应一个告警”的场景。

// 定义 PatternSelectFunction
DataStream<AlertResult> alertStream = patternStream.select(
    new PatternSelectFunction<LoginEvent, AlertResult>() {
        @Override
        public AlertResult select(Map<String, List<LoginEvent>> pattern) {
            // pattern 的 key 是模式名称，value 是匹配到的事件列表
            List<LoginEvent> first = pattern.get("first");
            LoginEvent lastEvent = first.get(first.size() - 1);
            return new AlertResult(
                lastEvent.getUserId(),
                "连续失败登录告警",
                System.currentTimeMillis()
            );
        }
    }
);

4.3.2 flatSelect：一对多输出

flatSelect() 接收 PatternFlatSelectFunction，允许一个匹配序列输出 零条、一条或多条 结果。适用于“一个匹配需要拆解为多个下游事件”的场景。

// 定义 PatternFlatSelectFunction
DataStream<AlertResult> alertStream = patternStream.flatSelect(
    new PatternFlatSelectFunction<LoginEvent, AlertResult>() {
        @Override
        public void flatSelect(
            Map<String, List<LoginEvent>> pattern,
            Collector<AlertResult> out
        ) {
            List<LoginEvent> events = pattern.get("first");
            // 对每个匹配事件单独输出告警
            for (LoginEvent event : events) {
                out.collect(new AlertResult(
                    event.getUserId(),
                    "单次失败登录告警",
                    event.getTimestamp()
                ));
            }
            // 额外输出一个汇总告警
            out.collect(new AlertResult(
                events.get(0).getUserId(),
                "连续失败登录汇总告警",
                System.currentTimeMillis()
            ));
        }
    }
);

4.3.3 select 与 flatSelect 对比

维度	select	flatSelect
输出数量	每个匹配序列输出 1 条	每个匹配序列输出 0 ~ N 条
函数签名	`PatternSelectFunction` → 返回单值	`PatternFlatSelectFunction` → 使用 Collector
适用场景	告警、聚合、单条记录输出	拆分、展开、多路输出
风控示例	“检测到盗刷，输出一条风控告警”	“检测到团伙欺诈，输出多条用户风险记录”

4.4 完整风控示例：实时登录风控

以下代码展示如何将 Pattern API、PatternStream 和 select 组合，实现“5分钟内同一用户连续失败登录3次”的实时风控规则。

// 1. 定义事件流
DataStream<LoginEvent> loginStream = env
    .addSource(new FlinkKafkaConsumer<>("login_topic", ...))
    .keyBy(LoginEvent::getUserId);

// 2. 定义 Pattern
Pattern<LoginEvent, LoginEvent> failPattern = Pattern
    .<LoginEvent>begin("fail_events")
    .where(event -> "FAIL".equals(event.getStatus()))
    .times(3)
    .consecutive()
    .within(Time.minutes(5));

// 3. 生成 PatternStream
PatternStream<LoginEvent> patternStream = CEP.pattern(loginStream, failPattern);

// 4. 提取告警
DataStream<RiskAlert> alertStream = patternStream.select(pattern -> {
    List<LoginEvent> fails = pattern.get("fail_events");
    LoginEvent lastFail = fails.get(fails.size() - 1);
    return new RiskAlert(
        lastFail.getUserId(),
        "连续失败登录",
        "HIGH",
        fails.stream().map(LoginEvent::getIp).collect(Collectors.toList())
    );
});

// 5. 输出告警到下游
alertStream.addSink(new AlertSink());

性能优化建议： 在风控场景中，建议对 PatternStream 进行 .keyBy() 分区，确保同一用户的匹配计算在同一个 Task Slot 中执行，避免跨网络传输带来的延迟和状态不一致。

4.5 本章小结

Pattern API 是规则定义层，通过 .where()、.times()、.within() 等方法构建事件序列约束。
连续性策略（.next()、.followedBy()、.followedByAny()）决定了事件之间的匹配关系，直接影响风控规则的准确性。
PatternStream 是 CEP 的中间产物，内部基于 NFA 状态机进行事件匹配。
select 适用于一对一输出，flatSelect 适用于一对多输出，根据业务需求选择合适的提取方式。