4. Flink CEP 核心 API：Pattern API 概览

Flink CEP 的核心是 Pattern API，它提供了一套声明式、流式风格的编程接口，用于定义复杂事件匹配规则。本章将系统讲解 Pattern API 的四大核心概念：Pattern 定义、Pattern 序列、Pattern 组合以及整体 API 概览。掌握这些基础，是构建实时风控规则引擎的第一步。

4.1 Pattern API 整体架构

Pattern API 采用 Builder 模式，通过链式调用逐步构建匹配规则。其核心组件包括：

4.2 Pattern 定义：单个事件的匹配条件

Pattern 定义是构建规则的最小单元，用于描述一个“事件”应该满足的条件。在风控场景中，每个 Pattern 对应一个可疑行为片段。

4.2.1 创建 Pattern

使用 Pattern.<T>begin("name") 创建初始 Pattern，其中泛型 T 是事件类型，name 是 Pattern 的唯一标识。

// 定义：交易金额超过 10000 的事件
Pattern<TransactionEvent, TransactionEvent> largeAmount =
    Pattern.<TransactionEvent>begin("large_transaction")
        .where(event -> event.getAmount() > 10000);

4.2.2 条件约束方法

4.3 Pattern 序列：连接多个事件

Pattern 序列定义了事件之间的时间顺序和邻近关系。在风控场景中，序列用于描述“先 A 后 B”的异常行为链条。

4.3.1 序列连接方法

4.3.2 序列构建示例

// 风控规则：登录失败 → 修改密码 → 大额转账（严格连续）
Pattern<LoginEvent, LoginEvent> riskPattern =
    Pattern.<LoginEvent>begin("login_fail")
        .where(e -> e.getStatus().equals("FAIL"))
        .next("change_pwd")
        .where(e -> e.getType().equals("PASSWORD_CHANGE"))
        .next("large_transfer")
        .where(e -> e.getAmount() > 50000);

4.4 Pattern 组合：逻辑与时间约束

Pattern 组合提供了逻辑运算符和时间窗口，用于构建更复杂的风控规则。

4.4.1 逻辑组合

• AND 组合：多个条件同时满足，通过 .where() 链式调用实现。
• OR 组合：多个条件满足其一，通过 .or() 方法实现。
• NOT 组合：否定某个事件的出现，通过 .notNext() 或 .notFollowedBy() 实现。
• 循环组合：事件重复出现，通过 .times(n)、.oneOrMore()、.timesOrMore(n) 实现。

4.4.2 时间约束

使用 .within(time) 为整个 Pattern 序列设置时间窗口，超过时间未匹配则丢弃。

// 风控规则：5分钟内，同一用户连续3次交易失败 → 触发告警
Pattern<TransactionEvent, TransactionEvent> failPattern =
    Pattern.<TransactionEvent>begin("fail1")
        .where(e -> e.getStatus().equals("FAIL"))
        .next("fail2")
        .where(e -> e.getStatus().equals("FAIL"))
        .next("fail3")
        .where(e -> e.getStatus().equals("FAIL"))
        .within(Time.minutes(5));  // 整个序列必须在5分钟内完成

4.4.3 循环模式与量词

4.5 完整风控 Pattern 示例

以下是一个综合使用 Pattern 定义、序列、组合的实时风控规则：

// 规则：10分钟内，同一用户从新设备登录 + 修改密码 + 大额转账（金额 > 20000）
Pattern<RiskEvent, RiskEvent> highRiskPattern =
    Pattern.<RiskEvent>begin("new_device_login")
        .where(e -> e.getType().equals("LOGIN"))
        .where(e -> e.isNewDevice())
        .followedBy("password_change")
        .where(e -> e.getType().equals("PWD_CHANGE"))
        .next("large_transfer")
        .where(e -> e.getType().equals("TRANSFER"))
        .where(e -> e.getAmount() > 20000)
        .within(Time.minutes(10));  // 整个序列必须在10分钟内完成

// 应用到 DataStream
DataStream<RiskEvent> inputStream = ...;
PatternStream<RiskEvent> patternStream = CEP.pattern(inputStream, highRiskPattern);

4.6 本章小结