2. Flink 环境搭建与项目初始化

本章将引导你完成从零搭建 Flink 开发环境、理解集群部署模式、创建 Maven/Gradle 项目并运行第一个 Flink 程序的全过程。这是后续所有批流一体特征工程案例的基础。

2.1 本地开发环境配置

推荐使用以下工具链进行 Flink 应用开发：

2.2 Flink 集群模式介绍

Flink 支持多种部署模式，适用于不同规模和场景。以下是三种核心模式的对比：

2.3 Maven/Gradle 项目搭建

我们提供两种主流构建工具的快速初始化方式。推荐使用 Maven 作为本课程的标准构建工具。

2.3.1 使用 Maven 创建项目

在终端执行以下命令，或使用 IntelliJ IDEA 的 Maven 项目向导：

mvn archetype:generate \
    -DarchetypeGroupId=org.apache.flink \
    -DarchetypeArtifactId=flink-quickstart-java \
    -DarchetypeVersion=1.17.1 \
    -DgroupId=com.featureeng \
    -DartifactId=flink-feature-engineering \
    -Dversion=1.0-SNAPSHOT \
    -Dpackage=com.featureeng \
    -DinteractiveMode=false

生成的项目结构如下：

flink-feature-engineering/
├── pom.xml
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── com/
│   │   │       └── featureeng/
│   │   │           ├── BatchJob.java
│   │   │           ├── StreamingJob.java
│   │   │           └── SocketWindowWordCount.java
│   │   └── resources/
│   │       └── log4j.properties
│   └── test/
│       └── java/
│           └── com/
│               └── featureeng/
│                   └── BatchJobTest.java

2.3.2 使用 Gradle 创建项目

对于偏好 Gradle 的开发者，在 build.gradle 中添加以下依赖：

plugins {
    id 'java'
    id 'application'
}

repositories {
    mavenCentral()
}

dependencies {
    implementation 'org.apache.flink:flink-java:1.17.1'
    implementation 'org.apache.flink:flink-streaming-java:1.17.1'
    implementation 'org.apache.flink:flink-clients:1.17.1'
}

application {
    mainClass = 'com.featureeng.StreamingJob'
}

2.4 第一个 Flink 程序：批流一体 WordCount

我们编写一个同时支持批处理和流处理的 WordCount 程序，展示 Flink 批流一体的核心能力。

2.4.1 批处理版本 (BatchJob.java)

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class BatchJob {
    public static void main(String[] args) throws Exception {
        // 1. 创建批处理执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取数据源
        DataSet<String> text = env.fromElements(
            "Hello Flink", "Hello World", "Flink Feature Engineering");

        // 3. 转换：分词 & 计数
        DataSet<Tuple2<String, Integer>> counts = text
            .flatMap(new Tokenizer())
            .groupBy(0)
            .sum(1);

        // 4. 输出结果
        counts.print();
    }

    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] tokens = value.toLowerCase().split("\\W+");
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}

2.4.2 流处理版本 (StreamingJob.java)

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

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

        // 2. 读取 socket 数据流 (需先启动 nc -lk 9999)
        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // 3. 转换：分词 & 计数
        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap(new Tokenizer())
            .keyBy(0)
            .sum(1);

        // 4. 输出结果
        counts.print();

        // 5. 执行作业
        env.execute("Streaming WordCount");
    }

    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] tokens = value.toLowerCase().split("\\W+");
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}

2.5 运行与验证

1. 批处理运行： 在 IDE 中直接运行 BatchJob.main()，控制台将输出单词统计结果。
2. 流处理运行：
   • 先启动 socket 数据源：nc -lk 9999
   • 再运行 StreamingJob.main()
   • 在 nc 终端输入文本，观察 Flink 控制台实时输出计数结果。

2.6 常见问题与避坑指南