2. 环境准备：Flink集群搭建（Standalone模式）、JDK配置、Maven项目初始化、常用依赖引入

本章是实战教程的基石。在开始坏样本归因分析之前，必须搭建一套可运行的 Flink 计算环境，并完成开发项目的初始化。我们将采用 Standalone 模式部署 Flink 集群，该模式不依赖 YARN 或 Kubernetes，适合开发测试与教学场景。

2.1 JDK 配置

Flink 1.13+ 要求 JDK 8 或 11。推荐使用 JDK 8u201+ 或 JDK 11.0.6+。配置步骤如下：

1. 下载 JDK（以 Linux 为例，解压至 /usr/local/java）。
2. 编辑 /etc/profile 或 ~/.bashrc，添加环境变量。
3. 执行 source 使配置生效，验证 java -version。

# 解压
tar -zxvf jdk-8u201-linux-x64.tar.gz -C /usr/local/java/

# 配置环境变量
export JAVA_HOME=/usr/local/java/jdk1.8.0_201
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

# 验证
java -version
# 输出示例：java version "1.8.0_201"

2.2 Flink Standalone 集群搭建

Standalone 模式由 JobManager（主节点）和 TaskManager（工作节点）组成。我们在一台物理机上模拟三节点集群（1 个 Master + 2 个 Worker）。

2.2.1 下载与解压

wget https://archive.apache.org/dist/flink/flink-1.14.6/flink-1.14.6-bin-scala_2.12.tgz
tar -zxvf flink-1.14.6-bin-scala_2.12.tgz -C /opt/
cd /opt/flink-1.14.6

2.2.2 核心配置

编辑 conf/flink-conf.yaml，设置以下关键参数：

2.2.3 配置 workers 文件

编辑 conf/workers，添加 TaskManager 节点地址（单机模拟时使用 localhost）：

localhost
localhost

若为多机集群，此处填写各 Worker 的主机名或 IP，每行一个。

2.2.4 启动与验证

# 启动集群
./bin/start-cluster.sh

# 查看进程
jps
# 应看到：StandaloneSessionClusterEntrypoint (JobManager)
#         TaskManagerRunner (TaskManager) × 2

# 访问 Web UI
# 浏览器打开 http://localhost:8081

2.3 Maven 项目初始化

使用 Maven 创建 Flink 项目骨架，推荐使用 flink-quickstart-java 原型（archetype）。

2.3.1 命令行创建

mvn archetype:generate \
  -DarchetypeGroupId=org.apache.flink \
  -DarchetypeArtifactId=flink-quickstart-java \
  -DarchetypeVersion=1.14.6 \
  -DgroupId=com.realtime \
  -DartifactId=flink-badsample-analysis \
  -Dversion=1.0-SNAPSHOT \
  -Dpackage=com.realtime.badsample \
  -DinteractiveMode=false

2.3.2 项目结构

初始化后的目录结构如下：

flink-badsample-analysis/
├── pom.xml
└── src/
    └── main/
        ├── java/
        │   └── com/realtime/badsample/
        │       ├── BatchJob.java
        │       └── StreamingJob.java
        └── resources/
            └── log4j.properties

2.4 常用依赖引入

编辑 pom.xml，添加坏样本分析所需的依赖。以下为最小依赖集：

2.4.1 pom.xml 核心片段

<properties>
    <flink.version>1.14.6</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
    <java.version>1.8</java.version>
</properties>

<dependencies>
    <!-- Flink Core -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- Table API & SQL -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- Connector -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-csv</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- Lombok -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.24</version>
        <scope>provided</scope>
    </dependency>

    <!-- Logging -->
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.2.11</version>
    </dependency>
</dependencies>

2.5 环境验证

完成上述步骤后，执行以下命令验证环境就绪：

# 1. 验证 JDK
java -version

# 2. 验证 Flink 集群
curl http://localhost:8081 | grep "Task Managers"
# 应返回包含 "2" 的 JSON 片段

# 3. 验证 Maven 编译
cd flink-badsample-analysis
mvn clean package -DskipTests
# 输出 BUILD SUCCESS