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

3. Flink CEP 核心API：Pattern API详解、select与flatSelect、超时事件处理

在风控稳定性巡检场景中，Flink CEP（Complex Event Processing）是识别复杂事件序列的核心引擎。本章深入解析Pattern API的三大支柱：模式定义（Pattern API）、结果提取（select/flatSelect）以及超时事件处理。掌握这些API是构建高精度风控规则的基础。

3.1 Pattern API：定义事件模式的基石

Pattern API是Flink CEP的声明式DSL，用于定义待检测的事件序列模式。一个完整的模式由个体模式（Individual Pattern）和组合条件构成。

3.1.1 个体模式与量词

个体模式定义单个事件的匹配条件，并通过量词控制事件出现的次数。

量词方法	语义	风控场景示例
`.times(n)`	精确匹配n次	同一设备在1分钟内登录失败恰好3次
`.times(n, m)`	匹配n到m次	同一IP在5分钟内触发风控规则2~5次
`.oneOrMore()`	匹配1次或多次（贪婪）	连续转账操作至少1次
`.timesOrMore(n)`	匹配n次或以上	同一收货地址下单超过3次
`.optional()`	该事件可选出现	登录后可能有修改密码操作

  巡检要点：量词使用不当会导致状态膨胀。例如.oneOrMore()在无界流中可能产生无限状态，建议配合.times(n, m)或.until()限制匹配次数。

3.1.2 条件定义：where与subtype

条件用于过滤事件，支持简单条件、组合条件以及子类型条件。

// 简单条件：基于事件属性过滤
.where(new SimpleCondition<LoginEvent>() {
    @Override
    public boolean filter(LoginEvent event) {
        return event.getStatus().equals("FAILED");
    }
})

// 组合条件：使用AND/OR逻辑
.where(new SimpleCondition<LoginEvent>() {
    @Override
    public boolean filter(LoginEvent event) {
        return event.getAttempts() > 3;
    }
}).or(new SimpleCondition<LoginEvent>() {
    @Override
    public boolean filter(LoginEvent event) {
        return event.getRegion().equals("high_risk");
    }
})

// 子类型条件：匹配特定子类事件
.subtype(HighRiskLoginEvent.class)

3.1.3 模式序列与连续性

多个个体模式通过连续性策略组合成序列，决定事件之间的邻近关系。

连续性策略	方法	含义	状态开销
严格连续	`.next()`	事件必须严格按顺序出现，中间不能有其他事件	低
宽松连续	`.followedBy()`	事件按顺序出现，允许中间有其他事件	中
非确定性宽松连续	`.followedByAny()`	允许事件重叠匹配，同一个事件可被多个模式使用	高
不允许出现	`.notNext()`	指定事件不能紧跟在前面事件之后	中

风控避坑：非确定性宽松连续（followedByAny）会显著增加状态存储和回溯计算量。在风控高吞吐场景下，优先使用followedBy，仅在需要检测重叠匹配（如滑动窗口内的重复攻击）时才使用followedByAny。

3.2 select与flatSelect：提取匹配结果

当模式匹配成功后，需要通过select或flatSelect将匹配的事件序列转换为输出结果。两者核心区别在于输出类型：select返回单条记录，flatSelect可返回多条记录。

3.2.1 select：单条结果输出

select接收一个PatternSelectFunction，将匹配的事件映射为单个输出对象。

// 定义模式：连续3次登录失败后，5分钟内发生转账
Pattern<LoginEvent, ?> fraudPattern = Pattern
    .<LoginEvent>begin("failures")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return event.getStatus().equals("FAILED");
        }
    })
    .times(3)
    .next("transfer")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return event.getType().equals("TRANSFER");
        }
    })
    .within(Time.minutes(5));

// 使用select提取结果
PatternStream<LoginEvent> patternStream = CEP.pattern(inputStream, fraudPattern);

DataStream<Alert> alertStream = patternStream.select(
    new PatternSelectFunction<LoginEvent, Alert>() {
        @Override
        public Alert select(Map<String, List<LoginEvent>> pattern) throws Exception {
            List<LoginEvent> failures = pattern.get("failures");
            LoginEvent transfer = pattern.get("transfer").get(0);

            return new Alert(
                transfer.getUserId(),
                "连续登录失败后转账",
                failures.size(),
                transfer.getAmount()
            );
        }
    }
);

3.2.2 flatSelect：多结果输出

flatSelect通过PatternFlatSelectFunction的collector参数，允许一次匹配输出零到多条记录。适用于一个模式匹配需要拆分为多个告警或事件的场景。

// 使用flatSelect：一次匹配输出多个告警
DataStream<Alert> alertStream = patternStream.flatSelect(
    new PatternFlatSelectFunction<LoginEvent, Alert>() {
        @Override
        public void flatSelect(
                Map<String, List<LoginEvent>> pattern,
                Collector<Alert> out) throws Exception {

            List<LoginEvent> failures = pattern.get("failures");
            LoginEvent transfer = pattern.get("transfer").get(0);

            // 输出高风险告警
            if (transfer.getAmount() > 10000) {
                out.collect(new Alert("HIGH", "大额转账风险", transfer));
            }

            // 输出设备指纹告警
            out.collect(new Alert("MEDIUM", "设备异常", failures.get(0)));

            // 输出行为序列告警
            out.collect(new Alert("LOW", "行为模式匹配", transfer));
        }
    }
);

特性	select	flatSelect
输出数量	严格1:1（一次匹配输出一条）	1:N（一次匹配输出0~N条）
接口类型	`PatternSelectFunction`	`PatternFlatSelectFunction`
典型场景	单一告警、聚合统计	多级告警、事件拆分、日志审计
性能开销	低	中（需处理Collector）

巡检建议：在风控稳定性巡检中，推荐优先使用flatSelect进行告警分级输出。一次模式匹配可以同时产生高风险阻断告警、中风险观察告警和低风险日志记录，避免为不同级别重复定义相似模式。

3.3 超时事件处理：捕获未完成序列

风控场景中，事件序列可能因超时而未完全匹配。Flink CEP通过within(Time)定义时间窗口，并支持PatternTimeoutFunction处理超时事件。

3.3.1 超时机制原理

当模式定义了within(Time)后，Flink CEP会为每个部分匹配的序列启动定时器。如果在时间窗口内未完成所有模式匹配，则触发超时回调。

// 定义带超时的模式：10分钟内完成"登录失败3次 → 转账"
Pattern<LoginEvent, ?> timeoutPattern = Pattern
    .<LoginEvent>begin("failures")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return event.getStatus().equals("FAILED");
        }
    })
    .times(3)
    .next("transfer")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return event.getType().equals("TRANSFER");
        }
    })
    .within(Time.minutes(10));  // 超时窗口

3.3.2 使用PatternTimeoutFunction处理超时

通过select或flatSelect的重载版本，同时传入超时处理函数和正常匹配处理函数。

// 超时处理：捕获未完成转账的登录失败序列
DataStream<Alert> timeoutAlertStream = patternStream.select(
    // 超时回调函数
    new PatternTimeoutFunction<LoginEvent, Alert>() {
        @Override
        public Alert timeout(
                Map<String, List<LoginEvent>> pattern,
                long timeoutTimestamp) throws Exception {

            List<LoginEvent> failures = pattern.get("failures");
            return new Alert(
                failures.get(0).getUserId(),
                "超时告警：登录失败后未完成转账",
                failures.size(),
                timeoutTimestamp
            );
        }
    },
    // 正常匹配回调函数
    new PatternSelectFunction<LoginEvent, Alert>() {
        @Override
        public Alert select(Map<String, List<LoginEvent>> pattern) throws Exception {
            List<LoginEvent> failures = pattern.get("failures");
            LoginEvent transfer = pattern.get("transfer").get(0);
            return new Alert(
                transfer.getUserId(),
                "正常匹配：登录失败后转账",
                failures.size(),
                transfer.getAmount()
            );
        }
    }
);

3.3.3 超时事件在风控中的价值

风控场景	超时事件含义	巡检动作
登录失败 → 转账	用户登录失败3次后，10分钟内未转账	降低风险等级，标记为"可疑但未完成"
批量注册 → 下单	同一设备注册多个账号后，1小时内未下单	封禁设备，防止后续批量攻击
修改密码 → 提现	修改密码后，30分钟内未提现	临时冻结提现功能，触发二次验证
多次查询 → 转账	频繁查询余额后，5分钟内未转账	记录行为日志，用于后续模型训练

3.4 综合示例：风控登录异常检测

将上述API整合为一个完整的风控规则：检测"同一用户10分钟内登录失败3次以上，且后续有转账行为"的异常模式，并处理超时情况。

// 完整风控规则实现
public class LoginFraudDetector {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1. 定义事件流
        DataStream<LoginEvent> inputStream = env
            .fromElements(
                new LoginEvent("user1", "FAILED", 1L),
                new LoginEvent("user1", "FAILED", 2L),
                new LoginEvent("user1", "FAILED", 3L),
                new LoginEvent("user1", "TRANSFER", 4L, 5000.0),
                new LoginEvent("user2", "FAILED", 5L),
                new LoginEvent("user2", "FAILED", 6L),
                new LoginEvent("user2", "FAILED", 7L)
                // user2 超时，未发生转账
            )
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
            );

        // 2. 定义模式
        Pattern<LoginEvent, ?> pattern = Pattern
            .<LoginEvent>begin("failures")
            .where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent event) {
                    return event.getStatus().equals("FAILED");
                }
            })
            .times(3)
            .next("transfer")
            .where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent event) {
                    return event.getType().equals("TRANSFER");
                }
            })
            .within(Time.minutes(10));

        // 3. 应用模式
        PatternStream<LoginEvent> patternStream = CEP.pattern(inputStream, pattern);

        // 4. 处理结果（含超时）
        DataStream<Alert> resultStream = patternStream.select(
            // 超时处理
            (PatternTimeoutFunction<LoginEvent, Alert>) (pattern, timestamp) -> {
                List<LoginEvent> failures = pattern.get("failures");
                return new Alert(
                    failures.get(0).getUserId(),
                    "TIMEOUT_ALERT",
                    "登录失败3次后超时未转账",
                    failures.size(),
                    0.0
                );
            },
            // 正常匹配
            (PatternSelectFunction<LoginEvent, Alert>) pattern -> {
                List<LoginEvent> failures = pattern.get("failures");
                LoginEvent transfer = pattern.get("transfer").get(0);
                return new Alert(
                    transfer.getUserId(),
                    "MATCH_ALERT",
                    "登录失败3次后发生转账",
                    failures.size(),
                    transfer.getAmount()
                );
            }
        );

        resultStream.print();

        env.execute("Login Fraud Detection");
    }
}

  巡检总结：Pattern API是Flink CEP的骨架，select/flatSelect是输出通道，超时处理是风控场景的必备能力。三者结合可以覆盖从简单阈值到复杂序列的各类风控规则。在稳定性巡检中，重点关注：量词是否导致状态膨胀、连续性策略是否合理、超时时间是否匹配业务SLA。