2. Flink 环境搭建与运行：本地模式安装、Standalone 集群搭建、YARN 模式部署、第一个 Flink 程序（WordCount）

环境搭建，说白了就是让 Flink 能在你的机器上跑起来。很多初学者觉得这步很枯燥，但我得说一句——环境搞不好，后面全是坑。我自己带团队时，至少有一半的排查时间都花在环境问题上。所以这一章，咱们把四种模式都过一遍，你跟着走就行。

2.1 本地模式安装：最快跑起来的方式

本地模式适合学习、调试、小规模测试。不需要 Hadoop，不需要 YARN，一台机器就够了。

2.1.1 前置条件

  • JDK 1.8 或 11（推荐 1.8，我踩过 JDK 11 某些版本的坑）
  • Linux / macOS / Windows（WSL 也行）
  • 网络能下载 Flink 包

2.1.2 安装步骤

  1. 从官网下载 Flink 1.17.0（当前稳定版）： 
    wget https://dlcdn.apache.org/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz
  2. 解压： 
    tar -xzf flink-1.17.0-bin-scala_2.12.tgz
cd flink-1.17.0
  3. 启动集群： 
    ./bin/start-cluster.sh
  4. 检查是否启动成功： 
    jps
# 应该看到 StandaloneSessionClusterEntrypoint 和 TaskManagerRunner
  5. 访问 Web UI：http://localhost:8081
我的习惯：每次启动后先看一眼 Web UI 的 Task Managers 页面。如果 Task Slot 数量为 0，说明 TaskManager 没注册成功，八成是端口冲突或内存不够。

2.1.3 验证安装

跑一个简单的命令测试一下：

./bin/flink run examples/streaming/WordCount.jar

看到控制台输出单词计数，就说明本地模式搞定了。

2.2 Standalone 集群搭建：多台机器一起干活

本地模式只能单机，生产环境肯定不够。Standalone 模式是 Flink 自带的集群管理方式，不依赖外部资源调度器。我个人觉得，Standalone 适合中小团队、实验环境，或者对资源隔离要求不高的场景。

2.2.1 集群架构

先看一张图，理解 Standalone 集群里各角色的关系：

Standalone 集群架构 JobManager TaskManager 1 Slot: 4 TaskManager 2 Slot: 4 TaskManager 3 Slot: 4 Client JobManager 负责调度，TaskManager 负责执行，Client 负责提交作业 每个 TaskManager 有多个 Slot，Slot 是资源分配的最小单位

2.2.2 搭建步骤

假设你有三台机器：node1（主节点）、node2、node3。

  1. 在所有节点上安装 JDK 并解压 Flink。
  2. 修改 conf/flink-conf.yaml（在 node1 上）： 
    jobmanager.rpc.address: node1
jobmanager.rpc.port: 6123
taskmanager.numberOfTaskSlots: 4
parallelism.default: 2
  3. 修改 conf/workers，添加 worker 节点： 
    node2
node3
  4. 将配置好的 Flink 目录分发到所有节点： 
    scp -r flink-1.17.0 node2:/opt/
scp -r flink-1.17.0 node3:/opt/
  5. 在 node1 上启动集群： 
    ./bin/start-cluster.sh
  6. 验证：访问 http://node1:8081，看 Task Managers 页面是否显示 node2 和 node3。
我曾经踩过的坑：忘记配置 SSH 免密登录，结果启动时一直报连接拒绝。还有一次，workers 文件里写了 localhost，结果所有 TaskManager 都跑在同一台机器上。嗯，这两个问题排查了我一下午。

2.3 YARN 模式部署：企业级首选

YARN 模式是生产环境最常用的部署方式。为什么？因为 YARN 能动态分配资源，跟 Hadoop 生态无缝集成。我在大厂做实时数仓时，所有 Flink 作业都跑在 YARN 上。

2.3.1 两种运行模式

模式 特点 适用场景
Session 模式 共享 JobManager，多个作业共用集群 短作业、测试、小规模
Per-Job 模式 每个作业独享 JobManager，资源隔离好 生产环境、长作业、重要作业

2.3.2 前置条件

  • Hadoop 集群已部署（YARN 可用）
  • 设置环境变量 HADOOP_CLASSPATH
  • Flink 包中已包含 Hadoop 依赖（或手动添加）

2.3.3 启动 Session 模式

# 启动 YARN Session
./bin/yarn-session.sh -n 4 -s 4 -jm 1024 -tm 2048

# 参数说明：
# -n  : TaskManager 数量
# -s  : 每个 TaskManager 的 Slot 数
# -jm : JobManager 内存 (MB)
# -tm : TaskManager 内存 (MB)

启动成功后，会输出一个 Application ID，比如 application_1690000000000_0001。你可以在 YARN 的 ResourceManager UI 上看到它。

2.3.4 提交作业到 YARN Session

./bin/flink run -m yarn-cluster -yid application_1690000000000_0001 \
  examples/streaming/WordCount.jar

2.3.5 使用 Per-Job 模式（推荐）

./bin/flink run -m yarn-cluster \
  -yjm 1024 -ytm 2048 \
  -ys 4 \
  examples/streaming/WordCount.jar

Per-Job 模式不需要先启动 Session，直接提交作业，YARN 会自动分配资源。作业结束后，资源自动释放。

我的建议：生产环境用 Per-Job 模式。Session 模式虽然省事，但多个作业挤在一起，一个作业出问题可能影响其他作业。我曾经因为 Session 模式下的内存泄漏，导致整个集群的作业都挂了。

2.4 第一个 Flink 程序：WordCount

WordCount 是 Flink 的 Hello World。别看它简单，里面包含了 Flink 编程的核心概念：Source、Transformation、Sink。

2.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 WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取数据源（这里用 socket 模拟实时流）
        DataStream<String> text = env.socketTextStream("localhost", 9999);

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

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

        // 5. 执行作业
        env.execute("Flink 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));
                }
            }
        }
    }
}

2.4.2 运行步骤

  1. 启动 netcat 发送数据： 
    nc -lk 9999
  2. 在 IDE 中运行 WordCount 程序，或者在命令行提交： 
    ./bin/flink run -c WordCount /path/to/WordCount.jar
  3. 在 netcat 窗口输入单词，比如： 
    hello world
hello flink
flink is awesome
  4. 观察控制台输出： 
    (hello,1)
(world,1)
(hello,2)
(flink,1)
(flink,2)
(is,1)
(awesome,1)

2.4.3 代码解析

  • StreamExecutionEnvironment：Flink 程序的入口，负责设置执行参数。
  • socketTextStream：从 socket 读取数据，模拟实时流。
  • flatMap：将一行文本切分成多个单词，每个单词输出一个 (word, 1) 对。
  • keyBy：按单词分组，相当于 SQL 里的 GROUP BY。
  • sum：对第二个字段求和，统计每个单词出现的次数。
  • print：将结果输出到标准输出，调试时很方便。
注意：Flink 是懒加载的，所有算子只是构建了一个执行计划，直到调用 env.execute() 才会真正开始执行。这个特性跟 Spark 的 Action 操作很像。

2.5 本章小结

这一章我们走完了 Flink 的四种部署模式：本地模式适合学习，Standalone 适合小集群，YARN 模式是生产首选。最后用 WordCount 程序跑通了第一个 Flink 作业。你想想看，从零到跑起来，其实没那么复杂吧？

← 上一章 📖 返回目录 下一章 →