第2章：Flink基础与环境搭建——架构、部署与开发环境配置

说实话，很多初学者一上来就急着写代码，结果连Flink怎么跑起来的都没搞明白。我个人习惯是，先花点时间把架构和部署环境摸透，后面写代码才不慌。这一章，我们就从Flink的架构设计讲起，再到集群部署的两种主流方式，最后把开发环境配好。

2.1 Flink架构：核心组件与数据流

Flink的架构，说白了就是一套「主从分工」的体系。我刚开始接触时，总觉得它跟Spark差不多，但用久了才发现，Flink在流处理上的设计更「原生」。

你想想看，一个作业提交后，JobManager先把它解析成执行图，然后分发给TaskManager。每个TaskManager里跑着多个Task Slot，这就是并行度的基本单位。

2.2 Flink集群部署：Standalone模式

Standalone模式，就是Flink自己管自己，不依赖外部资源调度器。适合开发测试，或者小规模集群。我在项目初期做原型验证时，就喜欢用这个模式，省心。

部署步骤：

1. 下载Flink安装包，解压到所有节点。
2. 配置 conf/flink-conf.yaml，设置JobManager地址。
3. 配置 conf/workers，列出所有TaskManager节点。
4. 在JobManager节点执行 bin/start-cluster.sh。

核心配置项，我整理了一张表，方便你对照着配：

配置项	默认值	说明
jobmanager.rpc.address	localhost	JobManager的RPC地址
jobmanager.rpc.port	6123	RPC通信端口
taskmanager.numberOfTaskSlots	1	每个TaskManager的Slot数
parallelism.default	1	作业默认并行度

2.3 Flink集群部署：YARN模式

YARN模式才是生产环境的主流。它把资源管理交给Hadoop YARN，Flink只负责计算。说白了，就是「借鸡生蛋」——YARN帮你管机器，Flink专心跑任务。

YARN模式下有两种提交方式：

• Session模式：先启动一个长集群，然后往里面提交作业。适合短作业、交互式查询。
• Per-Job模式：每个作业启动一个独立集群，作业结束集群自动销毁。适合长作业、资源隔离要求高的场景。

我记得有一次，团队里有人用Session模式跑多个作业，结果一个作业的内存泄漏把整个集群拖垮了。从那以后，我们规定：生产环境必须用Per-Job模式。

YARN模式部署要点：

1. 确保Hadoop集群已启动，YARN和HDFS都可用。
2. 设置环境变量 HADOOP_CLASSPATH，Flink需要用它找到YARN的配置。
3. 使用 bin/yarn-session.sh 启动Session集群，或直接用 bin/flink run -m yarn-cluster 提交Per-Job作业。

2.4 Flink开发环境配置

环境配好了，接下来就是搭开发环境。我个人习惯用IntelliJ IDEA + Maven，当然Gradle也行，看个人喜好。

步骤一：创建Maven项目

在pom.xml里引入Flink依赖。注意版本号要跟集群一致，否则跑起来会报序列化异常——我踩过这个坑。

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

<dependencies>
    <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</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

步骤二：配置日志框架

Flink默认用Log4j 2，但很多项目用的是Logback。我建议统一用Log4j 2，省得冲突。在 src/main/resources 下放一个 log4j2.properties：

rootLogger.level = INFO
rootLogger.appenderRef.console.ref = ConsoleAppender
appender.console.name = ConsoleAppender
appender.console.type = CONSOLE
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n

步骤三：编写第一个Flink程序

写一个最简单的WordCount，验证环境是否通。嗯，这里要注意，Flink的DataSet API在1.12以后已经标记为废弃，新项目一律用DataStream API。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.socketTextStream("localhost", 9999)
            .flatMap((String line, Collector<String> out) -> {
                for (String word : line.split(" ")) {
                    out.collect(word);
                }
            })
            .keyBy(word -> word)
            .sum(1)
            .print();
        
        env.execute("Socket WordCount");
    }
}

2.5 环境验证与常见问题

环境配好后，怎么验证？我的做法是三步走：

1. 启动Flink集群，访问Web UI（默认端口8081），确认JobManager和TaskManager都活着。
2. 提交一个简单的作业，比如上面的WordCount，看能不能正常跑完。
3. 检查TaskManager的日志，确认没有报错。

如果遇到问题，别慌。我列几个常见坑：

• 端口冲突：8081被占了？改 rest.port 配置。
• 类加载冲突：依赖版本不一致，尤其是Hadoop和Flink的guava版本。用 mvn dependency:tree 排查。
• 内存不足：YARN模式下，Container内存配小了。调大 taskmanager.memory.process.size。

好了，这一章的内容就到这里。架构、部署、开发环境，这三样东西是Flink的「地基」。地基打牢了，后面写实时特征计算的代码才稳当。