3. Flink CEP 环境搭建：Flink集群安装、Maven依赖配置、本地开发环境初始化

在开始复杂事件处理（CEP）的开发之前，搭建一个稳定、可复现的运行环境是首要任务。本章将引导你完成从零开始的Flink集群安装、Maven项目依赖配置以及本地开发环境的初始化，确保后续的CEP案例能够顺利运行。

3.1 环境概览与版本选型

本课程采用经过生产验证的稳定版本组合，兼顾功能完整性与社区活跃度。以下为推荐的技术栈版本：

3.2 Flink 集群安装（本地模式）

对于学习和开发调试，我们采用 Standalone 本地集群 模式。该模式无需依赖 Hadoop 或 YARN，启动快速，适合单机环境下的CEP逻辑验证。

3.2.1 下载与解压

# 下载 Flink 1.17.2 二进制包
wget https://archive.apache.org/dist/flink/flink-1.17.2/flink-1.17.2-bin-scala_2.12.tgz

# 解压到指定目录（例如 /opt/flink）
tar -xzf flink-1.17.2-bin-scala_2.12.tgz -C /opt/flink
cd /opt/flink/flink-1.17.2

3.2.2 启动本地集群

# 启动 Standalone 集群（默认使用 1 个 JobManager 和 1 个 TaskManager）
./bin/start-cluster.sh

# 验证进程是否启动
jps
# 应看到以下进程：
# 12345 StandaloneSessionClusterEntrypoint  (JobManager)
# 12346 TaskManagerRunner                  (TaskManager)

3.2.3 访问 Web UI

启动成功后，打开浏览器访问 http://localhost:8081。你将看到 Flink 的 Dashboard，其中可以查看作业状态、TaskManager 资源、任务执行计划等信息。

3.3 Maven 依赖配置

创建一个标准的 Maven 项目，并在 pom.xml 中引入 Flink 核心依赖以及 CEP 专用依赖。以下是经过精简且无冲突的依赖配置模板：

<properties>
    <flink.version>1.17.2</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
    <java.version>1.8</java.version>
</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>

    <!-- Flink CEP 复杂事件处理依赖 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-cep</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>

    <!-- 日志桥接（避免控制台无日志输出） -->
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.2.11</version>
    </dependency>
</dependencies>

3.4 本地开发环境初始化

完成依赖配置后，我们需要编写一个简单的 Flink 作业来验证环境是否就绪。该作业将读取一个本地 Socket 数据流，并打印到控制台。

3.4.1 创建主类

package com.flink.cep.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class EnvironmentCheckJob {
    public static void main(String[] args) throws Exception {
        // 1. 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 设置并行度（本地调试建议设为 1）
        env.setParallelism(1);

        // 3. 从本地 Socket 读取数据（需先启动 nc -lk 9999）
        DataStream<String> textStream = env.socketTextStream("localhost", 9999);

        // 4. 简单的转换：将每行文本拆分为单词
        DataStream<String> wordStream = textStream
                .flatMap((FlatMapFunction<String, String>) (line, out) -> {
                    for (String word : line.split("\\s+")) {
                        out.collect(word);
                    }
                })
                .returns(String.class);

        // 5. 打印结果
        wordStream.print("Flink-CEP-Env");

        // 6. 执行作业
        env.execute("Flink CEP Environment Check");
    }
}

3.4.2 启动 Socket 数据源并运行

# 终端1：启动数据源
nc -lk 9999

# 终端2：在 IDE 中运行 EnvironmentCheckJob 主类
# 或在命令行使用 Maven 运行：
mvn clean compile exec:java -Dexec.mainClass="com.flink.cep.demo.EnvironmentCheckJob"

在 nc 终端输入 hello flink cep，IDE 控制台应输出：

Flink-CEP-Env> hello
Flink-CEP-Env> flink
Flink-CEP-Env> cep

3.5 环境架构总览

下图展示了本地开发环境的核心组件及其交互关系，帮助你理解数据流从 Socket 到 Flink 集群再到控制台的完整路径。

3.6 常见问题与避坑指南

• 端口占用： 若 localhost:8081 无法访问，检查是否已有 Flink 进程在运行，或使用 lsof -i:8081 查看端口占用情况。
• 依赖冲突： 避免同时引入不同版本的 flink-cep 和 flink-streaming-java，使用 mvn dependency:tree 检查依赖树。
• 并行度设置： 本地调试时建议将 env.setParallelism(1)，避免多线程导致的日志混乱和调试困难。
• Socket 连接超时： 如果运行作业后立即报连接拒绝，请确认 nc -lk 9999 已提前启动。