文章目录

前言

Apache Flink作为领先的分布式流处理框架，其灵活的部署和运行模式是支撑各类实时计算场景的核心基础。本文档系统梳理了Flink的三种主要运行模式——Standalone、YARN与Kubernetes，并重点剖析了生产环境中应用最广泛的YARN集成方案。在此基础上，详细介绍了Flink集群的核心组件角色、三种部署模式（会话模式、单作业模式、应用模式）的差异与适用场景，以及历史服务器的配置与使用。通过对这些内容的掌握，读者能够根据实际业务需求选择合适的部署策略，高效构建和管理Flink集群。

一、Flink运行模式

（一）Standalone运行模式（了解）

  • 独立模式是独立运行的，不依赖任何外部的资源管理平台；
  • 当然独立也是有代价的：如果资源不足，或者出现故障，没有自动扩展或重分配资源的保证，必须手动处理。
  • 所以独立模式一般只用在开发测试或作业非常少的场景下。
    在这里插入图片描述

（二）YARN运行模式（重点）

  • YARN上部署的过程是：客户端把Flink应用提交给Yarn的ResourceManager，Yarn的ResourceManager会向Yarn的NodeManager申请容器。
  • 在这些容器上，Flink会部署JobManager和TaskManager的实例，从而启动集群。
  • Flink会根据运行在JobManger上的作业所需要的Slot数量动态分配TaskManager资源。

在这里插入图片描述

（三）K8S 运行模式（了解）

  • 容器化部署是如今业界流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。
  • 容器管理工具中最为流行的就是Kubernetes（k8s），而Flink也在最近的版本中支持了k8s部署模式。
  • 基本原理与YARN是类似的，具体配置可以参见官网说明，这里我们就不做过多讲解了。

二、Flink集群搭建

（一）集群规划

在这里插入图片描述

（二）集群角色

Flink提交作业和执行任务，需要几个关键组件：

1.客户端（Client）

代码由客户端获取并做转换，之后提交给JobManger

2.JobManager

Flink集群里的**“管事人”，对作业进行中央调度管理**；而它获取到要执行的作业后，会进一步处理转换，然后分发任务给众多的TaskManager。

3.TaskManager

真正“干活的人”，数据的处理操作都是它们来做的。

在这里插入图片描述

  • Flink是一个非常灵活的处理框架，它支持多种不同的部署场景，还可以和不同的资源管理平台方便地集成。所以接下来我们会先做一个简单的介绍，让大家有一个初步的认识，之后再展开讲述不同情形下的Flink部署。

三、部署模式

在一些应用场景中，对于集群资源分配和占用的方式，可能会有特定的需求。
Flink为各种场景提供了不同的部署模式，主要有以下三种：

  • 会话模式（Session Mode）
  • 单作业模式（Per-Job Mode）
  • 应用模式（Application Mode）。

它们的区别主要在于：集群的生命周期以及资源的分配方式；以及应用的main方法到底在哪里执行——客户端（Client）还是JobManager。

（一）会话模式（Session Mode）

1.概念

会话模式其实最符合常规思维。我们需要先启动一个集群，保持一个会话，在这个会话中通过客户端提交作业。集群启动时所有资源就都已经确定，所以所有提交的作业会竞争集群中的资源。

在这里插入图片描述
会话模式比较适合于单个规模小、执行时间短的大量作业。

2.部署步骤

YARN的会话模式需要首先申请一个YARN会话（YARN Session）来启动Flink集群。具体步骤如下：

（1）启动集群
  • 启动Hadoop集群（HDFS、YARN）。
  • 执行脚本命令向YARN集群申请资源，开启一个YARN会话，启动Flink集群。
[atguigu@hadoop102 flink-1.17.0]$ bin/yarn-session.sh -nm test

可用参数解读：
-d：分离模式，如果你不想让Flink YARN客户端一直前台运行，可以使用这个参数，即使关掉当前对话窗口，YARN session也可以后台运行。
-jm（–jobManagerMemory）：配置JobManager所需内存，默认单位MB。
-nm（–name）：配置在YARN UI界面上显示的任务名。
-qu（–queue）：指定YARN队列名。
-tm（–taskManager）：配置每个TaskManager所使用内存。

  • 注意：Flink1.11.0版本不再使用-n参数和-s参数分别指定TaskManager数量和slot数量，YARN会按照需求动态分配TaskManager和slot。所以YARN的会话模式不会把集群资源固定，是动态分配的。

  • YARN Session启动之后会给出一个Web UI地址以及一个YARN application ID，如下所示，用户可以通过Web UI或者命令行两种方式提交作业。
    通过运行后台输出日志，可以看到程序ID和对应的接口

2022-11-17 15:20:52,711 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface hadoop104:40825 of application 'application_1668668287070_0005'.
JobManager Web Interface: http://hadoop104:40825
（2）提交作业
①通过Web UI提交作业

在这里插入图片描述

②通过命令行提交作业
  • 将FlinkTutorial-1.0-SNAPSHOT.jar任务上传至集群。
  • 执行以下命令将该任务提交到已经开启的Yarn-Session中运行。
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run
-c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar

客户端可以自行确定JobManager的地址，也可以通过-m或者-jobmanager参数指定JobManager的地址，JobManager的地址在YARN Session的启动页面中可以找到。

  • 任务提交成功后，可在YARN的Web UI界面查看运行情况。hadoop103:8088。
    在这里插入图片描述
    从上图中可以看到我们创建的Yarn-Session实际上是一个Yarn的Application，并且有唯一的Application ID。

  • 也可以通过Flink的Web UI页面查看提交任务的运行情况，如下图所示。
    在这里插入图片描述

（二）单作业模式（Per-Job Mode）

1.概念

会话模式因为资源共享会导致很多问题，所以为了更好地隔离资源，我们可以考虑为每个提交的作业启动一个集群，这就是所谓的单作业（Per-Job）模式。

在这里插入图片描述

  • 作业完成后，集群就会关闭，所有资源也会释放。
  • 这些特性使得单作业模式在生产环境运行更加稳定，所以是实际应用的首选模式。
  • 需要注意的是，Flink本身无法直接这样运行，所以单作业模式一般需要借助一些资源管理框架来启动集群，比如YARN、Kubernetes（K8S）。

2.部署步骤

在YARN环境中，由于有了外部平台做资源调度，所以我们也可以直接向YARN提交一个单独的作业，从而启动一个Flink集群。

（1）执行命令提交作业。
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run -d -t yarn-per-job -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar
  • 注意：如果启动过程中报如下异常。

Exception in thread “Thread-5” java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration ‘classloader.check-leaked-classloader’.
at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders

  • 解决办法： 在flink的/opt/module/flink-1.17.0/conf/flink-conf.yaml配置文件中设置
[atguigu@hadoop102 conf]$ vim flink-conf.yaml
classloader.check-leaked-classloader: false
（2）在YARN的ResourceManager界面查看执行情况

在这里插入图片描述

  • 点击可以打开Flink Web UI页面进行监控，如下图所示：

在这里插入图片描述

（3）可以使用命令行查看或取消作业，命令如下
[atguigu@hadoop102 flink-1.17.0]$ bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY

[atguigu@hadoop102 flink-1.17.0]$ bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY <jobId>
  • 这里的application_XXXX_YY是当前应用的ID，是作业的ID。注意如果取消作业，整个Flink集群也会停掉。

（三）应用模式（Application Mode）

1.概念

前面提到的两种模式下，应用代码都是在客户端上执行，然后由客户端提交给JobManager的。但是这种方式客户端需要占用大量网络带宽，去下载依赖和把二进制数据发送给JobManager；加上很多情况下我们提交作业用的是同一个客户端，就会加重客户端所在节点的资源消耗。

  • 解决办法：不要客户端，直接把应用提交到JobManger上运行。而这也就代表着，我们需要为每一个提交的应用单独启动一个JobManager，也就是创建一个集群。这个JobManager只为执行这一个应用而存在，执行结束之后JobManager也就关闭了，这就是所谓的应用模式。

在这里插入图片描述

  • 会话模式与单作业模式，都是提交作业之后才创建集群；
  • 单作业模式是通过客户端来提交的，客户端解析出的每一个作业对应一个集群；
  • 应用模式下，是直接由JobManager执行应用程序的。

！！！！三种模式的区别比较

对比维度 会话模式（Session Mode） 单作业模式（Per-Job Mode） 应用模式（Application Mode）
集群生命周期 先启动集群保持会话，所有作业共享集群 为每个提交的作业单独启动一个集群，作业完成后集群关闭 为每个提交的应用单独启动一个 JobManager，执行结束后关闭
资源分配方式 集群启动时资源已确定，所有提交的作业竞争集群中的资源 每个作业独立占用资源，作业间资源隔离 每个应用独立占用资源，互不干扰
main 方法执行位置 客户端（Client）执行 客户端（Client）执行 JobManager 直接执行
适用场景 单个规模小、执行时间短的大量作业 生产环境首选，稳定性更优 希望消除客户端资源消耗的场景
集群创建时机 提交作业之前已存在集群 提交作业后才创建集群 提交应用后才创建集群
资源释放 作业完成后集群继续保留，资源不释放 作业完成后集群关闭，所有资源释放 应用执行结束后集群关闭，资源释放
依赖外部框架 可独立运行，也可在 YARN/K8S 上运行 需要借助 YARN、Kubernetes 等资源管理框架 需要借助 YARN、Kubernetes 等资源管理框架

2.部署步骤

应用模式同样非常简单，与单作业模式类似，直接执行flink run-application命令即可。

（1）命令行提交
  • 执行命令提交作业
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run-application -t yarn-application -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar
  • 在命令行中查看或取消作业
[atguigu@hadoop102 flink-1.17.0]$ bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY

[atguigu@hadoop102 flink-1.17.0]$ bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>
（2）上传HDFS提交

可以通过yarn.provided.lib.dirs配置选项指定位置，将flink的依赖上传到远程。

  • 上传flink的lib和plugins到HDFS上
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -mkdir /flink-dist
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -put lib/ /flink-dist
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -put plugins/ /flink-dist
  • 上传自己的jar包到HDFS
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -mkdir /flink-jars
[atguigu@hadoop102 flink-1.17.0]$ hadoop fs -put FlinkTutorial-1.0-SNAPSHOT.jar /flink-jars
（3）提交作业
[atguigu@hadoop102 flink-1.17.0]$ bin/flink run-application -t yarn-application	-Dyarn.provided.lib.dirs="hdfs://hadoop102:8020/flink-dist"	-c com.atguigu.wc.SocketStreamWordCount  hdfs://hadoop102:8020/flink-jars/FlinkTutorial-1.0-SNAPSHOT.jar

这种方式下，flink本身的依赖和用户jar可以预先上传到HDFS，而不需要单独发送到集群，这就使得作业提交更加轻量了。

四、历史服务器

（一）history server是什么

  • Flink提供了历史服务器，用来在相应的 Flink 集群关闭后查询已完成作业的统计信息。
  • 只有当作业处于运行中的状态，才能够查看到相关的WebUI统计信息。
  • 通过 History Server 我们才能查询这些已完成作业的统计信息，无论是正常退出还是异常退出。

（二）使用步骤

1.创建存储目录

	hadoop fs -mkdir -p /logs/flink-job

2.在 flink-config.yaml中添加如下配置

jobmanager.archive.fs.dir: hdfs://hadoop102:8020/logs/flink-job
historyserver.web.address: hadoop102
historyserver.web.port: 8082
historyserver.archive.fs.dir: hdfs://hadoop102:8020/logs/flink-job
historyserver.archive.fs.refresh-interval: 5000

3.启动历史服务器

	bin/historyserver.sh start

4.停止历史服务器

bin/historyserver.sh stop

5.在浏览器地址栏输入

http://hadoop102:8082 查看已经停止的 job 的统计信息

总结

Flink的Standalone模式简单轻量，适合开发测试；YARN模式借助资源管理平台实现动态资源分配与隔离，是生产环境的常用选择；K8S模式则顺应容器化趋势，提供更灵活的运维能力。在部署模式上，会话模式适合短小的批量作业但存在资源竞争风险，单作业模式为每个作业启动独立集群，稳定性更优，应用模式则消除了客户端的资源消耗，通过直接将应用提交至JobManager实现更高效的执行。对于作业的持久化监控，历史服务器能够保留已终止作业的统计信息，便于事后分析与审计。合理组合运行环境与部署模式，并配置历史服务，可以有效提升Flink作业的交付效率与可观测性。

# flink # 大数据
Logo
openEuler 社区

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目，面向数字基础设施四大核心场景（服务器、云计算、边缘计算、嵌入式），全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐

从0开始的操作系统（4）

🔥 星恒随风：个人主页❄️ 个人专栏：《指针合集》《C语言基础》《数据结构》《机器学习导论》《前端基础》《python基础》《C++从入门到入土》✨ 数据即知识，压缩即智能。

avatar openEuler 社区
cover

java 7 Java 7被小米彻底抛弃？澎湃OS 4重写内核，性能飙得你尖叫

avatar openEuler 社区
cover

Python入门PDF在手，零基础也能秒变编程高手

avatar openEuler 社区