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

3. Flink基础入门：架构、环境与第一个流处理程序

好，咱们进入正题。Flink 是什么？说白了，它就是目前业界处理「实时数据流」最主流的计算引擎之一。我最早接触 Flink 是在 2018 年，当时公司要从 Storm 迁移，我调研了一圈，最后选了 Flink。为什么？因为它把「流」当作一等公民，而不是像 Spark Streaming 那样把流拆成微批。这个设计哲学，决定了它的实时性天花板更高。

3.1 Flink 核心架构

先看一张图，把 Flink 的「骨架」搞清楚。你想想看，一个 Flink 集群跑起来，到底谁在干什么？

这张图里三个角色，我一个个说。

3.2 核心概念：JobManager、TaskManager、Slot

JobManager：集群的「大脑」。负责接收作业、生成执行计划、协调检查点。如果 TaskManager 挂了，它负责重新调度。我在生产环境遇到过 JobManager 单点故障，后来都配了 HA（高可用），用 ZooKeeper 做 leader 选举。
TaskManager：真正干活的「工人」。每个 TaskManager 是一个 JVM 进程，里面可以划分多个 Slot。Slot 是资源调度的最小单位，一个 Slot 可以跑一个线程。
Slot：说白了就是 TaskManager 里的一小块计算资源（CPU + 内存）。我建议你把 Slot 数量设成跟机器 CPU 核数一致，别贪多，否则线程切换开销会让你哭。

💡 我的经验：刚开始学 Flink 的人，容易把 Slot 和并行度搞混。记住一句话：Slot 是「能跑多少线程」的上限，并行度是「实际跑多少线程」的设定。并行度不能超过 Slot 总数。

3.3 开发环境搭建（Java / Scala）

环境搭建这事儿，说简单也简单，说坑也多。我踩过的坑，今天一并告诉你。

3.3.1 前置条件

组件	版本要求	备注
JDK	8 / 11（推荐 11）	Flink 1.13+ 对 JDK 11 支持更好
Maven	3.6+	用于构建项目
IDE	IntelliJ IDEA	Eclipse 也行，但 IDEA 对 Flink 更友好
Scala（可选）	2.12 / 2.13	如果用 Scala API 才需要

⚠️ 避坑指南：我曾经在 Mac 上直接用 Homebrew 装 Flink，结果版本跟 Maven 依赖对不上，折腾了半天。建议你手动下载 Flink 二进制包，解压到固定目录，环境变量配好 FLINK_HOME。这样最稳。

3.3.2 创建 Maven 项目

我个人习惯用 Maven archetype 快速生成骨架。你打开终端，执行：

mvn archetype:generate \
  -DarchetypeGroupId=org.apache.flink \
  -DarchetypeArtifactId=flink-quickstart-java \
  -DarchetypeVersion=1.17.0 \
  -DgroupId=com.realtime.feature \
  -DartifactId=flink-demo \
  -Dversion=1.0-SNAPSHOT

生成后，目录结构是这样的：

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

嗯，这里要注意：StreamingJob.java 是流处理入口，BatchJob.java 是批处理入口。我们主要用前者。

3.4 第一个 Flink 流处理程序

光说不练假把式。咱们写一个最简单的程序：从 socket 读取文本，每 5 秒统计一次单词出现次数。这就是经典的「WordCount」流式版。

3.4.1 代码实现

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. 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 从 socket 读取数据（本地测试用）
        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // 3. 转换：切分单词，计数 1
        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            .sum(1);

        // 4. 打印结果
        counts.print();

        // 5. 启动执行
        env.execute("Flink Streaming WordCount");
    }

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

3.4.2 运行步骤

启动 socket 数据源：在终端执行 nc -lk 9999（Mac/Linux）或安装 ncat（Windows）。
启动 Flink 集群（本地模式）：./bin/start-cluster.sh，然后访问 http://localhost:8081 看 Web UI。
提交作业：用 IDE 直接运行 main 方法，或者打成 jar 包用 flink run 提交。
输入数据：在 nc 窗口里敲几行英文，比如 hello world，再敲 hello flink。
看结果：IDE 控制台会打印类似 (hello,1) (world,1) (hello,2) (flink,1) 的输出。

🔑 关键点解析：

socketTextStream：这是 Flink 自带的 Source，适合本地调试。生产环境一般用 Kafka Source。
keyBy：把相同单词分到同一个分区。底层用的是 hash 分区策略。
sum(1)：对 Tuple2 的第二个字段做累加。这是 Flink 提供的便捷算子。
print()：Sink 到标准输出。生产环境要换成 Kafka Sink 或 JDBC Sink。

3.5 并行度与数据流图

你想想看，上面那个程序跑起来，Flink 内部是怎么执行的？它会生成一张「数据流图」，每个算子（Source、FlatMap、KeyBy、Sum、Sink）都是一个节点，节点之间通过「数据通道」连接。

💡 我的习惯：在开发阶段，我会在 env.setParallelism(2) 设置全局并行度，然后在 Web UI 上看每个算子的并行度是否合理。如果某个算子成了瓶颈，单独给它设并行度。比如 text.flatMap(new Tokenizer()).setParallelism(4)。

3.6 常见问题与排查

问题	原因	解决方法
连接 socket 超时	nc 没启动或端口被占用	检查端口：`lsof -i:9999`
作业提交后一直 PENDING	集群资源不足，Slot 不够	增加 TaskManager 或减少并行度
控制台没输出	print() 被缓冲或日志级别不对	检查 log4j 配置，改成 DEBUG 级别
中文乱码	socket 编码不一致	统一用 UTF-8，nc 加 `-C` 参数

⚠️ 我曾经踩过的坑：有一次我在生产环境用 print() 做调试，结果忘了删，导致大量日志打到 stdout，把磁盘撑爆了。记住：print() 只用于本地测试，上线前一定要换成正式的 Sink。

3.7 本章小结

这一章我们干了三件事：

搞清楚了 Flink 的架构：Client 提交、JobManager 调度、TaskManager 执行。
搭好了开发环境：JDK + Maven + IDEA，跑通了本地集群。
写了第一个流处理程序：从 socket 读数据，做 WordCount，打印结果。