📚 课程目录第4章 Flink CEP第一个案例

4. Flink CEP第一个案例

本章通过一个最简化的风控场景——检测单个异常登录事件，带你快速上手 Flink CEP 的核心开发流程。你将理解事件定义、Pattern API 的基本用法以及如何将匹配结果输出到下游。

4.1 案例场景与目标

假设我们有一个用户登录事件流，每条事件包含用户ID、登录IP、登录时间戳和事件类型。风控规则要求：如果某个用户在短时间内（例如5秒内）连续登录失败超过3次，则判定为异常登录，触发告警。

字段	类型	说明
userId	String	用户唯一标识
ip	String	登录来源IP
eventTime	Long	事件发生时间戳（毫秒）
eventType	String	登录结果：success / fail

  关键点：本案例只关注 单个事件 的匹配，即只要出现一次 eventType = "fail" 就立即输出告警。这是 CEP 最简单的用法，用于建立对 Pattern 定义和匹配流程的直观认识。

4.2 核心开发步骤

Flink CEP 的开发流程可以归纳为以下 4 个步骤：

定义事件类型：创建 POJO 或 Java Bean 表示登录事件。
定义匹配模式：使用 Pattern API 描述“单个失败登录事件”。
将模式应用到流：通过 CEP.pattern() 将 DataStream 转换为 PatternStream。
处理匹配结果：使用 PatternSelectFunction 或 PatternFlatSelectFunction 提取告警信息。

4.3 代码实现

4.3.1 定义事件 POJO

public class LoginEvent {
    private String userId;
    private String ip;
    private Long eventTime;
    private String eventType;

    // 必须提供无参构造方法
    public LoginEvent() {}

    public LoginEvent(String userId, String ip, Long eventTime, String eventType) {
        this.userId = userId;
        this.ip = ip;
        this.eventTime = eventTime;
        this.eventType = eventType;
    }

    // getter / setter 方法（省略）
    // toString() 方法（省略）
}

4.3.2 定义匹配模式

// 使用 Pattern API 定义：匹配单个 eventType = "fail" 的事件
Pattern<LoginEvent, LoginEvent> failPattern = Pattern
    .<LoginEvent>begin("fail_event")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) throws Exception {
            return "fail".equals(event.getEventType());
        }
    });

模式说明：这里使用了 begin("fail_event") 定义模式名称，where() 方法指定过滤条件。由于只匹配单个事件，不需要指定时间窗口或次数约束。

4.3.3 应用模式并处理结果

// 将模式应用到事件流
DataStream<LoginEvent> inputStream = ...; // 从 Kafka 或 Socket 获取
PatternStream<LoginEvent> patternStream = CEP.pattern(
    inputStream.keyBy(LoginEvent::getUserId),
    failPattern
);

// 处理匹配结果，输出告警
DataStream<String> alertStream = patternStream.select(
    new PatternSelectFunction<LoginEvent, String>() {
        @Override
        public String select(Map<String, List<LoginEvent>> pattern) throws Exception {
            LoginEvent failEvent = pattern.get("fail_event").get(0);
            return "ALERT: 用户 " + failEvent.getUserId()
                   + " 在 IP " + failEvent.getIp()
                   + " 发生异常登录！时间: " + failEvent.getEventTime();
        }
    }
);

alertStream.print(); // 打印告警到控制台

4.4 运行与验证

启动 Flink 作业后，向输入流发送以下测试数据：

userId	ip	eventTime	eventType	预期结果
user_001	192.168.1.10	1000	success	不触发告警
user_001	192.168.1.10	2000	fail	触发告警
user_002	10.0.0.5	3000	fail	触发告警
user_001	192.168.1.10	4000	success	不触发告警

注意事项：

由于使用了 keyBy(LoginEvent::getUserId)，模式匹配是按用户维度独立进行的。
单个事件匹配不涉及时间窗口，只要事件满足条件就会立即触发。
实际生产环境中，建议使用 PatternFlatSelectFunction 处理超时事件。

4.5 本章小结

通过本案例，你完成了 Flink CEP 的第一次实战：

掌握了事件 POJO 的定义规范（必须有无参构造）。
学会了使用 Pattern.begin().where() 定义简单匹配模式。
理解了 CEP.pattern() 与 PatternSelectFunction 的配合方式。
验证了单个事件匹配的触发逻辑。