4. 环境搭建与项目初始化：基于 Maven/Gradle 创建 Flink 项目

本章将引导你完成从零搭建 Flink 开发环境，并初始化一个可用于压力测试特征实时计算的标准项目。我们将分别介绍 Maven 和 Gradle 两种构建工具的项目创建方式，并详细配置本地开发环境与集群运行环境。

4.1 开发环境前置检查

在开始创建项目之前，请确保你的开发机器已安装以下基础软件，并确认版本符合要求。

4.2 基于 Maven 创建项目

Maven 是 Flink 官方推荐的主流构建工具。我们通过 maven-archetype-quickstart 快速生成项目骨架。

4.2.1 使用命令行生成项目

mvn archetype:generate \
    -DarchetypeGroupId=org.apache.flink \
    -DarchetypeArtifactId=flink-quickstart-java \
    -DarchetypeVersion=1.17.0 \
    -DgroupId=com.stresstest \
    -DartifactId=flink-stress-test \
    -Dversion=1.0-SNAPSHOT \
    -Dpackage=com.stresstest.feature \
    -DinteractiveMode=false

4.2.2 核心 pom.xml 配置

生成项目后，打开 pom.xml，确认并补充以下关键依赖。注意 <scope>provided</scope> 的用法，这是区分本地运行与集群打包的关键。

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

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

    <!-- Flink 客户端，用于本地执行 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>

    <!-- 压力测试特征计算常用 JSON 解析 -->
    <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>
</dependencies>

4.3 基于 Gradle 创建项目

如果你更习惯 Gradle，可以使用以下 build.gradle 配置。Gradle 的依赖声明更简洁，适合快速原型开发。

plugins {
    id 'java'
    id 'application'
}

group = 'com.stresstest'
version = '1.0-SNAPSHOT'

repositories {
    mavenCentral()
}

ext {
    flinkVersion = '1.17.0'
    scalaBinaryVersion = '2.12'
}

dependencies {
    implementation "org.apache.flink:flink-streaming-java:${flinkVersion}"
    implementation "org.apache.flink:flink-clients:${flinkVersion}"
    implementation "org.apache.flink:flink-connector-kafka:${flinkVersion}"

    runtimeOnly 'org.slf4j:slf4j-log4j12:1.7.36'
}

application {
    mainClass = 'com.stresstest.feature.StressTestJob'
}

// 打包时排除 Flink 自身依赖，避免冲突
task fatJar(type: Jar) {
    manifest {
        attributes 'Main-Class': application.mainClass
    }
    from {
        configurations.runtimeClasspath.collect { it.isDirectory() ? it : zipTree(it) }
    }
    with jar
    duplicatesStrategy = DuplicatesStrategy.EXCLUDE
}

4.4 项目目录结构

初始化完成后，标准的项目目录结构如下。请确保 resources 目录下包含日志配置文件。

flink-stress-test/
├── pom.xml                          # Maven 构建文件
├── build.gradle                     # Gradle 构建文件（二选一）
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── com/stresstest/feature/
│   │   │       ├── StressTestJob.java        # 主入口
│   │   │       ├── FeatureCalculator.java    # 特征计算逻辑
│   │   │       └── model/
│   │   │           └── MetricEvent.java      # 数据模型
│   │   └── resources/
│   │       ├── log4j.properties              # 日志配置
│   │       └── flink-conf.yaml               # 本地运行配置
│   └── test/
│       └── java/
│           └── com/stresstest/feature/
│               └── FeatureCalculatorTest.java
└── README.md

4.5 本地运行环境配置

在 IDE 中直接运行 Flink 作业时，需要配置 并行度 和 检查点 等参数。以下是一个典型的本地执行环境配置示例。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.configuration.Configuration;

public class StressTestJob {
    public static void main(String[] args) throws Exception {
        // 创建本地执行环境，并行度设为 2
        Configuration config = new Configuration();
        config.setInteger(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, 10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(2, config);

        // 启用检查点
        env.enableCheckpointing(10000);

        // 数据源与计算逻辑（后续章节展开）
        // env.addSource(...)
        //   .map(...)
        //   .print();

        env.execute("Flink Stress Test Feature Job");
    }
}

4.6 集群运行环境配置

将作业提交到 Flink 集群（Standalone / YARN / Kubernetes）时，需要打包为 fat jar（包含所有非 provided 依赖），然后通过 flink run 命令提交。

4.6.1 打包命令

# Maven 打包
mvn clean package -DskipTests

# Gradle 打包
gradle fatJar

4.6.2 提交到集群

# 提交到 Standalone 集群
./bin/flink run -m yarn-cluster -p 4 \
    -c com.stresstest.feature.StressTestJob \
    ./target/flink-stress-test-1.0-SNAPSHOT.jar

# 提交到 YARN Session 集群
./bin/flink run -m yarn-cluster -yjm 1024 -ytm 2048 \
    -c com.stresstest.feature.StressTestJob \
    ./target/flink-stress-test-1.0-SNAPSHOT.jar

4.7 环境配置对比总结

4.8 环境搭建流程总览

下图展示了从开发环境准备到集群提交的完整流程，帮助你建立全局视角。

4.9 常见问题与避坑指南

• 依赖冲突： 集群环境中 Flink 自带的 flink-shaded-hadoop 可能与你的 fat jar 中的 Hadoop 依赖冲突。解决方案：在打包时排除 org.apache.hadoop 相关依赖。
• 序列化异常： 自定义 POJO 类必须实现 Serializable 接口，并包含无参构造器，否则 Flink 的序列化器会抛出 InvalidTypesException。
• 并行度设置： 本地调试时并行度建议设为 1 或 2，便于观察日志；集群生产环境根据数据量和资源合理设置，通常为 TaskManager 槽位数量的 1-2 倍。
• 检查点超时： 本地文件系统做检查点可能因磁盘 I/O 瓶颈导致超时，建议集群环境使用 HDFS 或 S3 作为检查点存储后端。