3. 开发环境搭建：JDK、Maven、Flink集群与第一个WordCount程序

本章将引导你从零搭建一套完整的Flink流处理开发环境，涵盖JDK、Maven、Flink集群（Standalone与YARN模式）以及IDE的配置，并通过经典的WordCount程序验证环境可用性。环境版本选择遵循生产级稳定性原则，避免使用过新或过旧的组件。

3.1 环境版本矩阵与选型说明

以下为本次教程推荐的组件版本及其选型理由，建议严格对齐以避免兼容性问题：

组件 推荐版本 选型理由
JDK 1.8.0_202 (8u202) 或 11.0.18+ Flink 1.14+ 对JDK 8/11支持最稳定；JDK 8建议使用最后免费商用版本8u202
Maven 3.6.3 或 3.8.8 3.6.x/3.8.x 与Flink shade插件兼容性最佳；3.9+ 可能触发依赖解析异常
Flink 1.14.6 或 1.17.2 1.14为长期维护版，1.17为最新稳定版；两者均支持Standalone与YARN部署
Hadoop (YARN) 3.2.4 或 3.3.6 Flink on YARN需要Hadoop客户端；3.2.x/3.3.x 均通过官方认证
IDE IntelliJ IDEA 2023.2+ 社区版免费且对Maven/Scala/Flink插件支持完善
版本对齐提示： 若使用Flink 1.17，建议搭配JDK 11 + Maven 3.8.8；若使用Flink 1.14，JDK 8 + Maven 3.6.3 是经过大量生产验证的组合。

3.2 JDK与Maven安装配置

3.2.1 JDK安装与环境变量

  1. 下载JDK： 从Oracle Archive或Adoptium获取JDK 8u202或JDK 11.0.18的tar.gz包。
  2. 解压至指定目录： sudo tar -zxf jdk-8u202-linux-x64.tar.gz -C /usr/local/
  3. 配置环境变量： 编辑 /etc/profile 或 ~/.bashrc，添加以下内容：
export JAVA_HOME=/usr/local/jdk1.8.0_202
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
  1. 验证安装： 执行 java -version，输出应显示 1.8.0_202 或 11.0.18。

3.2.2 Maven安装与仓库配置

  1. 下载Maven： 从Apache官网获取 apache-maven-3.8.8-bin.tar.gz。
  2. 解压并配置环境变量：
export MAVEN_HOME=/usr/local/apache-maven-3.8.8
export PATH=$MAVEN_HOME/bin:$PATH
  1. 配置阿里云镜像（国内加速）： 修改 $MAVEN_HOME/conf/settings.xml 的 <mirrors> 段：
<mirror>
  <id>aliyunmaven</id>
  <mirrorOf>central</mirrorOf>
  <name>阿里云公共仓库</name>
  <url>https://maven.aliyun.com/repository/public</url>
</mirror>
  1. 验证： mvn -version 输出应包含Maven版本及Java版本信息。

3.3 Flink集群部署：Standalone与YARN模式

Flink支持多种部署模式，本教程重点覆盖Standalone（单机/集群）和YARN（资源托管）两种模式。下图展示了两种模式的架构差异：

Flink部署模式对比：Standalone vs YARN Standalone 模式 JobManager (Master) Standalone ResourceManager TaskManager 1 TaskManager 2 TaskManager N 特点：进程级管理，不依赖外部资源调度器 适用：开发测试、小规模生产 YARN 模式 YARN ResourceManager YARN NodeManager Flink JobManager YARN Container TaskManager (Slot) 特点：资源由YARN统一管理，动态分配Container 适用：生产环境、多租户集群

3.3.1 Standalone模式部署步骤

  1. 下载Flink： wget https://archive.apache.org/dist/flink/flink-1.14.6/flink-1.14.6-bin-scala_2.12.tgz
  2. 解压： tar -zxf flink-1.14.6-bin-scala_2.12.tgz -C /opt/
  3. 配置JobManager： 编辑 /opt/flink-1.14.6/conf/flink-conf.yaml，设置：
jobmanager.rpc.address: localhost
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 1600m
taskmanager.memory.process.size: 2048m
taskmanager.numberOfTaskSlots: 4
parallelism.default: 2
  1. 启动集群： /opt/flink-1.14.6/bin/start-cluster.sh
  2. 验证： 访问 http://localhost:8081 查看Web UI，确认JobManager与TaskManager均处于Running状态。

3.3.2 YARN模式部署配置

  1. 确保Hadoop环境就绪： hadoop version 输出正常，且 YARN ResourceManager 已启动。
  2. 配置Flink on YARN： 设置环境变量 export HADOOP_CLASSPATH=$(hadoop classpath)，并将此命令加入 ~/.bashrc。
  3. 提交Flink会话到YARN：
/opt/flink-1.14.6/bin/yarn-session.sh -n 2 -tm 2048 -s 2 -d

参数说明：-n 2 启动2个TaskManager；-tm 2048 每个TaskManager分配2GB内存；-s 2 每个TaskManager分配2个Slot；-d 后台运行。

  1. 验证： 通过YARN ResourceManager Web UI（默认 http://localhost:8088）查看Flink会话的Application状态。
YARN模式常见问题： 若提交时提示 ClassNotFoundException: org.apache.hadoop.yarn.conf.YarnConfiguration，请检查 HADOOP_CLASSPATH 是否完整包含所有Hadoop依赖jar包。

3.4 IDE配置与Maven项目骨架

3.4.1 IntelliJ IDEA配置

  1. 安装Scala插件： 如果使用Scala API，需在 Settings → Plugins 中搜索并安装 Scala 插件。
  2. 配置JDK： File → Project Structure → SDKs，添加JDK 8或11。
  3. 配置Maven： Settings → Build, Execution, Deployment → Build Tools → Maven，设置Maven home path为 /usr/local/apache-maven-3.8.8，并勾选 Always update snapshots。

3.4.2 创建Flink Maven项目

使用Maven Archetype快速生成项目骨架：

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

生成后的项目结构如下：

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

3.5 第一个Flink程序：流式WordCount

以下代码实现一个完整的流式WordCount程序，从socket读取文本流，实时统计单词出现次数：

package com.realtime.feature.wordcount;

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读取数据流（nc -lk 9999）
        DataStream<String> textStream = env.socketTextStream("localhost", 9999);

        // 3. 转换：切分单词 -> 映射为 (word, 1) -> 按word分组 -> 累加计数
        DataStream<Tuple2<String, Integer>> wordCounts = textStream
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            .sum(1);

        // 4. 打印结果到控制台
        wordCounts.print();

        // 5. 执行任务
        env.execute("Streaming WordCount");
    }

    // 自定义FlatMapFunction：将每行文本切分为单词，输出 (word, 1)
    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.6 程序运行与验证

3.6.1 启动数据源

在终端启动netcat作为数据源：

nc -lk 9999

3.6.2 提交作业到Flink集群

Standalone模式：

mvn clean package -DskipTests
/opt/flink-1.14.6/bin/flink run target/flink-wordcount-1.0-SNAPSHOT.jar

YARN模式：

mvn clean package -DskipTests
/opt/flink-1.14.6/bin/flink run -m yarn-cluster -yjm 1024 -ytm 2048 target/flink-wordcount-1.0-SNAPSHOT.jar

3.6.3 验证结果

在netcat终端输入以下文本：

hello flink
hello world
flink streaming wordcount

Flink TaskManager的控制台输出应类似：

(hello,1)
(flink,1)
(hello,2)
(world,1)
(flink,2)
(streaming,1)
(wordcount,1)
验证要点： 注意 hello 和 flink 的计数随着新行输入而递增，说明流处理状态正确维护了累加器。若计数始终为1，请检查 keyBy 逻辑或并行度设置。

3.7 环境搭建检查清单

使用下表快速确认所有组件是否就绪：

检查项 验证命令/操作 预期结果
JDK java -version 显示 1.8.0_202 或 11.0.18
Maven mvn -version 显示 3.6.3 或 3.8.8
Flink Standalone jps 或 Web UI :8081 StandaloneSessionClusterEntrypoint + TaskManagerRunner
Flink YARN yarn application -list 显示 Flink 会话 Application
WordCount 作业 Flink Web UI → Jobs 作业状态为 RUNNING，输出正确计数
← 上一章 📖 返回目录 下一章 →