📚 课程目录第4章 Flink CEP开发环境搭建

4. Flink CEP 开发环境搭建：Maven 项目配置、依赖引入、本地运行模式与集群模式

本章将指导你从零搭建一个完整的 Flink CEP 开发环境。我们将依次完成 Maven 项目骨架创建、核心依赖引入、本地 IDE 运行调试，以及最终将作业提交至 Flink 集群的完整流程。掌握这些基础操作，是后续实现实时交易风控规则的前提。

4.1 开发环境全景概览

在开始编码之前，先明确整个开发与运行环境的层次关系。下图展示了从开发机到 Flink 集群的典型部署链路：

4.2 Maven 项目骨架与核心依赖

推荐使用 Maven 作为项目构建工具。以下是一个标准的 pom.xml 配置，包含 Flink 基础依赖、CEP 库以及连接器依赖。

4.2.1 基础 pom.xml 配置

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.risk.cep</groupId>
    <artifactId>flink-cep-trading-risk</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <properties>
        <flink.version>1.17.1</flink.version>
        <scala.binary.version>2.12</scala.binary.version>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <!-- Flink Java 核心 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Flink CEP 核心库 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-cep</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Kafka 连接器（用于生产环境数据源） -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- 日志框架 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.36</version>
            <scope>runtime</scope>
        </dependency>

        <!-- Lombok（可选，简化 POJO） -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.28</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals><goal>shade</goal></goals>
                        <configuration>
                            <transformers>
                                <transformer implementation=
                                    "org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.risk.cep.TradingRiskJob</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

依赖作用域说明： 集群模式下，Flink 核心库（flink-java、flink-streaming-java、flink-cep）应设为 provided，避免与集群自带的 Flink 库冲突。连接器（如 Kafka）和自定义代码则需打包进 JAR。

4.2.2 依赖版本与作用域速查表

依赖名称	推荐版本	作用域	说明
flink-java	1.17.1	provided	Flink DataSet API（批处理）
flink-streaming-java	1.17.1	provided	Flink DataStream API（流处理核心）
flink-cep	1.17.1	provided	CEP 模式匹配引擎
flink-clients	1.17.1	provided	客户端提交工具
flink-connector-kafka	1.17.1	compile	Kafka 数据源/输出
slf4j-log4j12	1.7.36	runtime	日志实现
lombok	1.18.28	provided	简化 POJO 代码

4.3 本地运行模式：IDE 调试

本地模式使用 Flink 内置的 MiniCluster，无需额外部署。在 IDE 中直接运行主类即可启动作业，非常适合 CEP 规则逻辑的快速验证。

4.3.1 本地模式核心代码模板

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

public class TradingRiskJobLocal {
    public static void main(String[] args) throws Exception {
        // 1. 创建本地执行环境（自动使用 MiniCluster）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);  // 本地并行度建议设为 CPU 核数

        // 2. 模拟数据源（此处用 Socket 或集合）
        DataStream<TransactionEvent> inputStream = env
            .socketTextStream("localhost", 9999)
            .map(line -> parseEvent(line));

        // 3. 定义 CEP 模式（示例：连续大额交易）
        Pattern<TransactionEvent, ?> pattern = Pattern
            .<TransactionEvent>begin("first")
            .where(event -> event.getAmount() > 10000)
            .next("second")
            .where(event -> event.getAmount() > 10000);

        // 4. 应用模式并输出告警
        CEP.pattern(inputStream, pattern)
            .select(patternResult -> {
                // 生成告警
                return "高风险交易对: " + patternResult.toString();
            })
            .print();

        // 5. 启动执行
        env.execute("Flink CEP Local Trading Risk Detection");
    }
}

  本地调试技巧：
  使用 env.setParallelism(1) 可简化调试时的输出顺序。
数据源优先使用 socketTextStream 或 fromElements，避免依赖外部系统。
在 IDE 中设置断点，可逐步跟踪 CEP 模式匹配状态。

4.3.2 本地运行步骤

启动数据源： 在终端执行 nc -lk 9999（Linux/Mac）或使用 Netcat 工具。
运行主类： 在 IDE 中右键 TradingRiskJobLocal → Run。
输入测试数据： 在终端输入交易事件，例如 1001,alice,12000。
观察输出： IDE 控制台会打印匹配到的告警信息。

4.4 集群模式：作业打包与提交

当 CEP 规则验证通过后，需要将作业打包为 Fat JAR 并提交到 Flink 集群。集群模式支持 Standalone、YARN、Kubernetes 等部署方式。

4.4.1 打包 Fat JAR

使用 maven-shade-plugin 将依赖（如 Kafka 连接器）与业务代码合并为一个 JAR 包：

# 在项目根目录执行
mvn clean package -DskipTests

# 打包成功后，目标目录生成：
# target/flink-cep-trading-risk-1.0-SNAPSHOT.jar

4.4.2 提交至 Standalone 集群

# 将 JAR 提交到 Flink 集群
./bin/flink run \
  -m yarn-cluster \                    # 或 standalone: jobmanager-host:8081
  -c com.risk.cep.TradingRiskJob \
  -p 4 \                               # 并行度
  /path/to/flink-cep-trading-risk-1.0-SNAPSHOT.jar

4.4.3 本地模式 vs 集群模式对比

对比维度	本地模式	集群模式
运行环境	IDE 内嵌 MiniCluster	独立 Flink 集群（Standalone/YARN/K8s）
并行度	默认 1，可手动设置	根据集群资源动态分配
数据源	Socket、集合、本地文件	Kafka、Pulsar、HDFS 等生产系统
状态后端	MemoryStateBackend	RocksDB / FsStateBackend
适用阶段	规则开发、单元测试	生产上线、压力测试
作业管理	IDE 停止即结束	Web UI / CLI 管理，支持 Savepoint

4.5 常见问题与避坑指南

依赖冲突： 集群模式下，如果 JAR 中打包了 Flink 核心库（如 flink-streaming-java），会与集群自带的版本冲突，导致 NoClassDefFoundError 或 IllegalStateException。务必使用 provided 作用域。

本地运行报错 "ClassNotFoundException"： 检查 pom.xml 中是否遗漏了 flink-clients 依赖。
集群提交后作业立即失败： 查看 JobManager 日志，通常是因为 mainClass 配置错误或 JAR 包损坏。
CEP 模式不触发： 确认事件时间语义与水位线设置正确；本地调试时建议使用 ProcessingTime 简化问题。
Kafka 连接器版本不匹配： Flink 1.17 对应 Kafka 3.x 客户端，使用 flink-connector-kafka:1.17.1。

4.6 本章小结

通过本章学习，你已掌握：

使用 Maven 搭建 Flink CEP 项目并引入核心依赖。
在 IDE 中本地运行 CEP 作业进行规则调试。
打包 Fat JAR 并提交至 Flink 集群运行。
区分本地模式与集群模式的差异及适用场景。