📚 课程目录第3章 Flink基础入门

3、Flink基础入门：Flink架构、DataStream API、执行环境搭建、第一个Flink实时程序

好，咱们正式开始啃 Flink 这块硬骨头。

这一章是地基，我尽量讲得实在些。你想想看，做实时特征工程，Flink 几乎是绕不开的选项。为什么？因为它天生就是干这个的——处理无界数据流，低延迟，高吞吐，还有强大的状态管理能力。我最早接触 Flink 是在 1.4 版本，那时候社区还没现在这么火，但一上手我就知道，这玩意儿能成。

3.1 Flink 核心架构：从 JobManager 到 TaskManager

先看一张图，把 Flink 的运行时架构刻在脑子里。

这张图我建议你多看两眼。Flink 的架构其实不复杂，核心就三个角色：Client、JobManager、TaskManager。

Client：负责把写好的 Flink 程序打包成 JobGraph，提交给 JobManager。它不参与计算，提交完就可以退出了。
JobManager：整个集群的“大脑”。它负责调度任务、协调 Checkpoint、故障恢复。我遇到过新手把 JobManager 配成单点，结果一挂全完蛋。生产环境一定要配高可用。
TaskManager：真正干活的“工人”。每个 TaskManager 有多个 Slot，Slot 是资源调度的最小单位。一个 Slot 可以跑一个 SubTask（子任务）。

关键概念：并行度与 Slot

并行度决定了你的算子有几个并行实例。Slot 是 TaskManager 上固定的资源槽。我见过有人把并行度设成 100，但 Slot 只有 10，结果任务排队排到天荒地老。记住：并行度 ≤ 总 Slot 数，这是铁律。

3.2 DataStream API：流处理的编程模型

DataStream API 是 Flink 的核心抽象。说白了，就是把数据流当成一个集合，你可以对它做各种转换操作。

我习惯把 DataStream API 的编程模型总结成三步：

获取执行环境：创建 StreamExecutionEnvironment。
接入数据源：从 Kafka、Socket、文件等读取数据。
定义转换逻辑：用 map、filter、keyBy、window 等算子处理数据。
输出结果：写入 MySQL、Redis、Kafka 或打印到控制台。
触发执行：调用 env.execute()。

来看一个最基础的例子，从 Socket 读取文本，做单词计数：

public class SocketWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 接入数据源：从 Socket 读取
        DataStream<String> textStream = env.socketTextStream("localhost", 9999);

        // 3. 转换逻辑：切分、计数
        DataStream<Tuple2<String, Integer>> wordCounts = textStream
            .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
                for (String word : line.split(" ")) {
                    out.collect(new Tuple2<>(word, 1));
                }
            })
            .returns(Types.TUPLE(Types.STRING, Types.INT))
            .keyBy(value -> value.f0)
            .sum(1);

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

        // 5. 触发执行
        env.execute("Socket WordCount");
    }
}

避坑指南：我曾经在 flatMap 里忘了调用 returns() 指定返回类型，结果 Flink 在运行时抛类型擦除异常。Java 的泛型擦除是个坑，Flink 需要你显式告诉它类型。用 TypeInformation 或者 returns() 都能解决。

3.3 执行环境搭建：本地模式与集群模式

搭建 Flink 执行环境，我分两种情况讲：本地开发和线上部署。

3.3.1 本地模式（开发调试）

本地模式最简单，适合写代码调试。你只需要在 pom.xml 里加依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>1.17.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>1.17.0</version>
</dependency>

然后在 IDE 里直接跑 main 方法。Flink 会自动启动一个嵌入式的 JobManager 和 TaskManager，默认并行度等于 CPU 核数。

注意：本地模式默认并行度是电脑 CPU 核数。我有一台 16 核的 MacBook，跑本地测试时开了 16 个并行，结果 IDE 直接卡死。建议本地调试时手动设置并行度：env.setParallelism(2)。

3.3.2 Standalone 集群模式（线上部署）

生产环境我推荐用 Standalone 或者 YARN/K8S 模式。这里先讲 Standalone：

组件	配置项	说明
JobManager	jobmanager.rpc.address	指定 JobManager 的 IP 地址
TaskManager	taskmanager.numberOfTaskSlots	每个 TaskManager 的 Slot 数量，默认 1
并行度	parallelism.default	全局默认并行度，可在 flink-conf.yaml 中设置
Checkpoint	state.backend	推荐用 RocksDB，支持大状态

启动命令很简单：

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

# 提交任务
./bin/flink run -c com.example.SocketWordCount ./target/my-job.jar

# 查看任务列表
./bin/flink list

嗯，这里要注意：Standalone 模式适合小规模集群。如果公司有 Hadoop 集群，我建议用 YARN 模式，资源管理更灵活。

3.4 第一个 Flink 实时程序：从零到跑通

光说不练假把式。咱们手写一个完整的实时程序，从 Socket 读取数据，做简单的特征统计。

场景是这样的：模拟用户点击流，每来一条数据，统计当前用户的点击次数。这在实时特征里很常见——比如统计用户最近 5 分钟的点击量。

public class UserClickCount {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // 2. 数据源：Socket，格式 "user_id,click_type"
        DataStream<String> source = env.socketTextStream("localhost", 9999);

        // 3. 转换：解析成 POJO
        DataStream<UserClick> clicks = source
            .map(line -> {
                String[] fields = line.split(",");
                return new UserClick(fields[0], fields[1], System.currentTimeMillis());
            });

        // 4. 按用户分组，统计点击次数
        DataStream<Tuple2<String, Long>> countStream = clicks
            .map(click -> new Tuple2<>(click.userId, 1L))
            .returns(Types.TUPLE(Types.STRING, Types.LONG))
            .keyBy(value -> value.f0)
            .sum(1);

        // 5. 输出
        countStream.print();

        // 6. 执行
        env.execute("User Click Count");
    }

    // POJO 类
    public static class UserClick {
        public String userId;
        public String clickType;
        public long timestamp;

        public UserClick() {}

        public UserClick(String userId, String clickType, long timestamp) {
            this.userId = userId;
            this.clickType = clickType;
            this.timestamp = timestamp;
        }
    }
}

运行步骤：

启动 Netcat：nc -lk 9999
在 IDE 中运行 UserClickCount
在 Netcat 窗口输入：u001,click 回车
观察控制台输出：(u001,1)
再输入一次：u001,click，输出变为 (u001,2)

看到没？Flink 是有状态的。它记住了每个用户的点击次数，这就是状态管理的基础。我在做实时特征时，经常用这种模式来维护用户的累计值、均值、最大值等统计量。

个人经验：刚开始学 Flink 时，我总喜欢把所有逻辑写在一个大 map 里。后来发现这样调试特别痛苦。建议把每个转换步骤拆成独立的算子，配合 print() 打印中间结果，排查问题快得多。

好了，这一章的内容就这些。Flink 的架构、API、环境搭建、第一个程序，都是后面做实时特征工程的基石。你把这些跑通了，后面讲窗口、状态、容错时，上手会快很多。