4. Flink CEP 快速入门：环境搭建、第一个CEP程序与运行流程

本章是Flink CEP实战的起点。我们将从零搭建开发环境，编写第一个基于简单序列匹配的CEP程序，并拆解其代码结构与运行流程。掌握这些基础后，后续的复杂模式与阈值调优才能落地。

4.1 环境搭建：最小化依赖与版本选型

Flink CEP作为Flink的专用库，无需额外安装集群，只需在项目中引入正确的Maven/Gradle依赖即可。推荐版本组合如下：

Maven 核心依赖示例：

<properties>
    <flink.version>1.17.2</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-cep</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

4.2 第一个CEP程序：简单序列匹配

我们以“连续两笔大额交易”作为检测场景：同一账户在5分钟内发生两笔金额超过10000的交易，则触发告警。这是CEP中最基础的严格连续（Strict Contiguity）模式。

4.2.1 数据模型与源

// 交易事件POJO
public class TransactionEvent {
    public String accountId;
    public double amount;
    public long timestamp;

    public TransactionEvent() {}

    public TransactionEvent(String accountId, double amount, long timestamp) {
        this.accountId = accountId;
        this.amount = amount;
        this.timestamp = timestamp;
    }

    @Override
    public String toString() {
        return "TransactionEvent{" +
                "accountId='" + accountId + '\'' +
                ", amount=" + amount +
                ", timestamp=" + timestamp +
                '}';
    }
}

4.2.2 定义CEP模式

import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.windowing.time.Time;

// 定义模式：严格连续的两笔大额交易
Pattern<TransactionEvent, ?> pattern = Pattern
    .<TransactionEvent>begin("first")
        .where(new SimpleCondition<TransactionEvent>() {
            @Override
            public boolean filter(TransactionEvent event) {
                return event.amount > 10000;
            }
        })
    .next("second")   // 严格连续：second必须紧跟在first之后
        .where(new SimpleCondition<TransactionEvent>() {
            @Override
            public boolean filter(TransactionEvent event) {
                return event.amount > 10000;
            }
        })
    .within(Time.minutes(5));  // 整体时间窗口

4.2.3 应用模式并输出告警

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

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

        // 模拟数据源（实际替换为Kafka/Socket）
        DataStream<TransactionEvent> inputStream = env
            .fromElements(
                new TransactionEvent("acc_001", 15000, 1000L),
                new TransactionEvent("acc_001", 20000, 1001L),  // 匹配
                new TransactionEvent("acc_002", 8000,  1002L),
                new TransactionEvent("acc_001", 5000,  1003L)
            )
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<TransactionEvent>forMonotonousTimestamps()
                    .withTimestampAssigner((event, ts) -> event.timestamp)
            );

        // 将模式应用到流上
        DataStream<String> alertStream = CEP.pattern(inputStream.keyBy(e -> e.accountId), pattern)
            .select(new PatternSelectFunction<TransactionEvent, String>() {
                @Override
                public String select(Map<String, List<TransactionEvent>> pattern) throws Exception {
                    TransactionEvent first = pattern.get("first").get(0);
                    TransactionEvent second = pattern.get("second").get(0);
                    return String.format("[ALERT] 账户 %s 连续大额交易: %.2f -> %.2f",
                            first.accountId, first.amount, second.amount);
                }
            });

        alertStream.print();

        env.execute("Flink CEP QuickStart");
    }
}

4.3 代码结构与运行流程拆解

一个完整的Flink CEP程序由以下五个核心环节构成，每个环节对应一段独立代码逻辑：

1. 获取执行环境 — StreamExecutionEnvironment.getExecutionEnvironment()
2. 构建数据源 — 从Kafka、Socket或本地集合读取事件流，并分配时间戳与水位线
3. 定义模式 — 使用 Pattern.begin().next().where().within() 链式API描述事件序列规则
4. 应用模式检测 — CEP.pattern(stream, pattern) 生成 PatternStream
5. 处理匹配结果 — 通过 .select() 或 .flatSelect() 提取匹配事件并输出告警

4.4 模式匹配语义对比

CEP中“连续”有多种语义，初学者最容易混淆。下表对比了三种常用连续策略：

4.5 本章小结

通过本章，你已掌握：

• Flink CEP的最小依赖配置与版本选型
• 基于 Pattern.next() 的严格连续序列匹配
• CEP程序的五步运行流程：环境→数据源→模式→检测→结果
• 三种连续策略的语义差异与选型建议