📚 课程目录第3章 Flink CEP基础

3. Flink CEP 基础：概念、事件定义与模式 API

Flink CEP（Complex Event Processing，复杂事件处理）是 Flink 上层构建的声明式规则匹配引擎。它允许用户在无界数据流中定义事件组合模式，当输入事件序列满足预定义规则时，触发告警、聚合或下游输出。本章从 CEP 核心概念出发，逐步拆解事件定义、模式 API 及两个经典示例：连续事件匹配与超时事件处理。

3.1 CEP 核心概念

概念	说明	类比（规则引擎）
事件（Event）	数据流中的最小单元，携带时间戳与业务字段	规则引擎中的“事实”
模式（Pattern）	定义事件组合的规则模板，支持连续、循环、可选等语义	规则条件组
模式序列（Pattern Sequence）	多个模式按时间或逻辑顺序连接形成的匹配链	规则链
匹配（Match）	输入事件序列满足模式序列后产生的输出结果	规则命中
时间窗口（Time Window）	模式匹配的有效时间范围，超出则丢弃或触发超时	规则有效期

关键理解：CEP 的核心是将“事件流”转化为“模式匹配流”。每个匹配结果代表一次规则命中，可直接用于告警、评分或下游动作。

3.2 事件定义

在 Flink CEP 中，事件通常定义为 POJO 或 Java Bean，必须包含时间戳字段（用于事件时间语义）和业务标识字段。以下是一个典型的交易事件定义：

// 交易事件 POJO
public class TradeEvent {
    private String tradeId;      // 交易ID
    private String userId;       // 用户ID
    private double amount;       // 交易金额
    private String type;         // 交易类型：DEPOSIT / WITHDRAW / TRANSFER
    private long timestamp;      // 事件时间戳（毫秒）

    // 必须提供无参构造、getter/setter
    public TradeEvent() {}

    // 全参构造、getter/setter 省略
}

事件定义要点：

时间戳字段：用于事件时间处理，需配合 assignTimestampsAndWatermarks 使用。
业务键字段：如 userId，用于按用户分组匹配（keyBy）。
无参构造器：Flink 序列化机制要求 POJO 必须有无参构造。

3.3 模式（Pattern）API

Flink CEP 提供 Pattern<T, F> 类构建模式序列。API 分为三部分：

API 类别	方法	语义
单模式定义	`.begin("name")`	定义模式序列的起始模式
条件过滤	`.where(condition)` `.or(condition)`	为模式添加事件过滤条件，支持迭代条件
序列连接	`.next("name")` `.followedBy("name")` `.followedByAny("name")`	严格连续、非严格连续、非确定性连续
量词控制	`.times(n)` `.oneOrMore()` `.optional()`	控制事件出现次数
时间约束	`.within(time)`	定义模式匹配的有效时间窗口

  ⚠ 注意：.next() 要求事件严格连续（中间不能有其他事件）；.followedBy() 允许中间出现不匹配事件；.followedByAny() 允许跳过任意事件，且会生成多个匹配。

3.4 简单模式示例：连续事件匹配

场景：检测同一用户在 5 分钟内连续发生“存款 → 转账 → 提现”三笔交易，且每笔金额均大于 1000。

// 1. 定义模式序列
Pattern<TradeEvent, ?> pattern = Pattern
    .<TradeEvent>begin("deposit")
        .where(e -> "DEPOSIT".equals(e.getType()) && e.getAmount() > 1000)
    .next("transfer")
        .where(e -> "TRANSFER".equals(e.getType()) && e.getAmount() > 1000)
    .next("withdraw")
        .where(e -> "WITHDRAW".equals(e.getType()) && e.getAmount() > 1000)
    .within(Time.minutes(5));

// 2. 在 DataStream 上应用 CEP
DataStream<TradeEvent> inputStream = ...;  // 已 keyBy(userId) 并分配时间戳
PatternStream<TradeEvent> patternStream = CEP.pattern(inputStream, pattern);

// 3. 输出匹配结果
DataStream<String> alertStream = patternStream.process(
    (PatternSelectFunction<TradeEvent, String>) patternMap -> {
        TradeEvent deposit = patternMap.get("deposit").get(0);
        TradeEvent transfer = patternMap.get("transfer").get(0);
        TradeEvent withdraw = patternMap.get("withdraw").get(0);
        return String.format("连续交易告警 | 用户:%s | 存款:%.2f → 转账:%.2f → 提现:%.2f",
            deposit.getUserId(), deposit.getAmount(),
            transfer.getAmount(), withdraw.getAmount());
    }
);

执行逻辑：当同一 userId 的事件流中，按顺序出现 DEPOSIT → TRANSFER → WITHDRAW 且均在 5 分钟内，则触发一次匹配。中间若出现其他类型事件，.next() 会导致匹配失败。

3.5 超时事件处理

场景：检测同一用户在 10 分钟内发生“登录 → 支付”事件，但支付事件在 10 分钟内未到达，则触发超时告警。

// 1. 定义模式：登录后等待支付，超时时间 10 分钟
Pattern<TradeEvent, ?> timeoutPattern = Pattern
    .<TradeEvent>begin("login")
        .where(e -> "LOGIN".equals(e.getType()))
    .next("payment")
        .where(e -> "PAYMENT".equals(e.getType()))
    .within(Time.minutes(10));

// 2. 使用 PatternTimeoutFunction 处理超时
PatternStream<TradeEvent> timeoutPatternStream = CEP.pattern(inputStream, timeoutPattern);

DataStream<String> timeoutAlertStream = timeoutPatternStream.process(
    new PatternTimeoutFunction<TradeEvent, String>() {
        @Override
        public String timeout(Map<String, List<TradeEvent>> patternMap,
                              long timeoutTimestamp) throws Exception {
            TradeEvent login = patternMap.get("login").get(0);
            return String.format("⏰ 超时告警 | 用户:%s | 登录时间:%d | 超时时间戳:%d",
                login.getUserId(), login.getTimestamp(), timeoutTimestamp);
        }
    },
    new PatternSelectFunction<TradeEvent, String>() {
        @Override
        public String select(Map<String, List<TradeEvent>> patternMap) {
            TradeEvent login = patternMap.get("login").get(0);
            TradeEvent payment = patternMap.get("payment").get(0);
            return String.format("正常支付 | 用户:%s | 登录→支付间隔:%dms",
                login.getUserId(), payment.getTimestamp() - login.getTimestamp());
        }
    }
);

超时处理要点：
1. PatternTimeoutFunction 在超时发生时被调用，此时只包含已匹配的部分事件（如 login）。
2. PatternSelectFunction 仅在完整匹配时被调用。
3. 两个输出流可以合并或分别处理，实现“正常匹配”与“超时告警”双路输出。

3.6 本章小结

CEP 本质：将事件流转化为模式匹配流，每个匹配代表一次规则命中。
事件定义：POJO 需包含时间戳、业务键和无参构造。
模式 API：.begin() → .where() → .next()/.followedBy() → .within() 构成完整模式链。
连续事件：使用 .next() 实现严格顺序匹配，适合强依赖场景。
超时事件：.within() + PatternTimeoutFunction 实现超时告警，双路输出分离正常与异常。