Flink-运行组件和框架了解

￡神魔★判官ぃ 2022-09-07 15:51 137阅读 0赞

# Flink运行框架 #

## Flink 运行时的组件 ##

Flink 运行时架构主要包括四个不同的组件,它们会在运行流处理应用程序时协同工作:**作业管理器(JobManager)、资源管理器(ResourceManager)、任务管理器(TaskManager),以及分发器(Dispatcher)**。因为 Flink 是用 Java 和 Scala 实现的,所以所有组件都会运行在Java 虚拟机上。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]

每个组件的职责如下:

**作业管理器(JobManager)**

控 制 一 个 应 用 程 序 执 行的 主 进 程 , 也 就 是 说 ,每 个 应 用 程 序 都 会 被 一个 不 同 的JobManager 所控制执行。

JobManager 会先接收到要执行的应用程序,这个应用程序会包括:作业图(JobGraph)、逻辑数据流图(logical dataflow graph)和打包了所有的类、库和其它资源的 JAR 包。

JobManager 会把 JobGraph 转换成一个物理层面的数据流图,这个图被叫做“执行图”(ExecutionGraph),包含了所有可以并发执行的任务。

JobManager 会向资源管理器(ResourceManager)请求执行任务必要的资源,也就是**任务管理器(TaskManager)上的插槽(slot)**。一旦它获取到了足够的资源,就会将执行图分发到真正运行它们的TaskManager 上。

而在运行过程中,JobManager 会负责所有需要中央协调的操作,比如说检查点(checkpoints)的协调。

负责管理整个作业的角色。

**任务管理器(TaskManager)**

Flink 中的工作进程。通常在 Flink 中会有多个 TaskManager 运行,每一个 TaskManager都包含了一定数量的插槽(slots)。**插槽的数量限制了 TaskManager 能够执行的任务数量**。

启动之后,TaskManager 会向资源管理器注册它的插槽;收到资源管理器的指令后,TaskManager 就会将一个或者多个插槽提供给 JobManager 调用。JobManager 就可以向插槽分配任务(tasks)来执行了。

在执行过程中,一个 TaskManager 可以跟其它运行同一应用程序的 TaskManager 交换数据。

负责具体执行任务的。

**资源管理器(ResourceManager)**

主要负责管理任务管理器(TaskManager)的插槽(slot),TaskManger 插槽是 Flink 中定义的处理资源单元。

Flink 为不同的环境和资源管理工具提供了不同资源管理器,比如YARN、 Mesos、 K8s,以及 standalone 部署。

当 JobManager 申请插槽资源时, ResourceManager会将有空闲插槽的 TaskManager 分配给 JobManager。如果 ResourceManager 没有足够的插槽来满足 JobManager 的请求,它还可以向资源提供平台发起会话,以提供启动 TaskManager进程的容器。另外,ResourceManager 还负责终止空闲的 TaskManager,释放计算资源。

负责进行资源的调配。

**分发器(Dispatcher)**

可以跨作业运行,它为应用提交提供了 REST 接口。

当一个应用被提交执行时,分发器就会启动并将应用移交给一个 JobManager。由于是 REST 接口,所以 Dispatcher 可以作为集群的一个 HTTP 接入点,这样就能够不受防火墙阻挡。

Dispatcher 也会启动一个 Web UI,用来方便地展示和监控作业执行的信息。

Dispatcher 在架构中可能并不是必需的,这取决于应用提交运行的方式。

## 任务提交流程 ##

当一个应用提交执行时,Flink 的各个组件是如何交互协作的，以下是job模式：  
![在这里插入图片描述][0072d11d48c1473eb6d46c6b5aa7fc21.png]

Yarn模式任务提交流程

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]

1、Flink 任务提交后,Client 向 HDFS 上传 Flink 的 Jar 包和配置

2、向 Yarn-ResourceManager 提 交 任 务

3、ResourceManager 分 配 Container 资 源 并 通 知 对 应 的NodeManager 启动 ApplicationMaster, ApplicationMaster 启动后加载 Flink 的 Jar 包和配置构建环境,然后启动 JobManager

4、ApplicationMaster 向 ResourceManager申 请 资 源 启 动 TaskManager ,

5、ResourceManager 分 配 Container 资 源 后 , 由ApplicationMaster 通 知 资 源 所 在 节 点 的 NodeManager 启 动 TaskManager

6、NodeManager 加载 Flink 的 Jar 包和配置构建环境并启动 TaskManager

7、TaskManager启动后向 JobManager 发送心跳包,并等待 JobManager 向其分配任务。

## 任务调度原理 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 2]

客户端不是运行时和程序执行的一部分,但它用于准备并发送dataflow(JobGraph)给 Master(JobManager),然后 ,客户端断开连接或者维持连接以等待接收计算结果。

当Flink 集 群 启 动 后 , 首 先 会 启 动 一 个JobManger 和 一 个 或 多 个 的TaskManager。由 Client 提交 任务给 JobManager, JobManager 再 调度任 务到各 个TaskManager 去执 行,然 后 TaskManager 将 心跳 和统 计信息 汇报给 JobManager。TaskManager 之间以流的形式进行数据的 传输。上述三者均为独立的 JVM 进程。

Client 为提交 Job 的客户端,可以是运行在任何机器上(与 JobManager 环境连通即可)。提交 Job 后,Client 可以结束进程(Streaming 的任务),也可以不结束并等待结果返回。

JobManager 主 要 负 责 调 度 Job 并 协 调 Task 做 checkpoint 。从 Client 处接收到 Job 和 JAR 包等资源后,会生成优化后的执行计划,并以 Task 的单元调度到各个 TaskManager 去执行。

TaskManager 在启动的时候就设置好了槽位数( Slot),每个 slot 能 启动一个Task, Task 为 线 程 。 从 JobManager 处 接 收 需 要 部 署 的 Task, 部 署 启 动 后 , 与 自己的上游建立 Netty 连接,接收数据并处理。

实现并行计算的方法：

对每一个任务对设置对应的并行度，然后根据并行度进行任务的拆分，最后分配到不同的slot去执行（多线程）。

并行的任务，需要占用多少slot：

任务中并行度最高的那个并行度就是需要占用的slot。

### TaskManger 与 Slots ###

Flink 中每一个 **worker(TaskManager)都是一个 JVM 进 程,它可能会在独立 的线程 上执行一个或多个 subtask**。为了控制一个 worker 能接收多少个 task,worker 通过 task slot 来进行控制(一个 worker 至少有一个 task slot)。

每个 task slot 表示 TaskManager 拥有资源的**一个固 定大小的 子集** 。假如一个TaskManager 有三个 slot,那么它会将其管理的内存分成三份给各个 slot。资源 slot化意味着一个 subtask 将不需要跟来自其他 job 的 subtask 竞争被管理的内存,取而代之的是它将拥有一定数量的内存储备。需要注意的是,这里不会涉及到 CPU 的隔离,slot 目前仅仅用来隔离 task 的受管理的内存，但是推荐使用CPU的核数去设置slot的个数，这样的目的是为了一个slot占用一个CPU，不用和别的slot共享。

通过调整 task slot 的数 量,允许用户定义 subtask 之间如何互相隔离。如果一个TaskManager 一个 slot, 那将意味着每个 task group 运行在独立的 JVM 中(该 JVM可能是通过一个特定的容器启动的),而一个 TaskManager 多个 slot 意味着更多的subtask 可以共享同一个 JVM。而在同一个 JVM 进程中的 task 将共享 TCP 连接(基于多路复用)和心跳 消 息。它们也可能共享 数 据集和数据结构,因 此 这减少了每个task 的负载。

TaskManager与slot

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 3]

Flink 中每一个 TaskManager 都是一个JVM进程,它可能会在独立的线程上执行一个或多个子任务

为了控制一个 TaskManager 能接收多少个 task, TaskManager 通过 taskslot 来进行控制(一个 TaskManager 至少有一个 slot)

子任务的共享slot

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 4]

默认情况下,Flink 允许子任务共享 slot,即使它们是不同任务的子任务(前提是它们来自同一个 job)。 这样的结果是,一个 slot 可以保存作业的整个管道。

**Task Slot 是 静态的概 念**,**是 指 TaskManager 具有的 并发执行能力** ,可以通过参数 taskmanager.numberOfTaskSlots 进行配置;而并**行度 parallelism 是 动态 概念**,**即 TaskManager 运行 程序时 实际使用的并发能力** ,可以通过参数 parallelism.default进行配置。

前后发生的不同的任务才能共享同一个slot，同一个任务不能放到同一个slot当中，必须放到不同的slot才能做到并行执行。

也就是说,假设一共有 3 个 TaskManager,每一个 TaskManager 中的分配 3 个TaskSlot,也就是每个 TaskManager 可以接收 3 个 task,一共 9 个 TaskSlot,如果我们设置 parallelism.default=1,即运行程序默认的并行度为 1,9 个 TaskSlot 只用了 1个,有 8 个空闲,因此,设置合适的并行度才能提高效率。

并行子任务的分配

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 7]

### 程序与数据流(DataFlow) ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 8]

所有的 Flink 程序都是由三部分组成的: Source 、Transformation 和 Sink。

Source 负责读取数据源,Transformation 利用各种算子进行处理加工,Sink 负责输出。

在运行时,Flink 上运行的程序会被映射成“逻辑数据流”(dataflows),它包含了这三部分。

**每一个 dataflow 以一个 或多个 sources 开 始以一个 或多个 sinks 结束** 。dataflow 类似于任意的有向无环图(DAG)。

在大部分情况下,程序中的转换运算(transformations) 跟 dataflow 中的算子(operator)是一一对应的关系,但有时候,一个 transformation 可能对应多个 operator。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 9]

### 执行图(ExecutionGraph) ###

由 Flink 程序直接映射成的数据流图 是 StreamGraph,也被称为逻辑流图,因为它们表示的是计算逻辑的高级视图。为了执行一个 流处理程序,Flink 需要将逻辑流图转换为物理数据流图(也叫执行图),详细说明程序的执行方式。

Flink 中的执行图可以分成四层: StreamGraph -> JobGraph -> ExecutionGraph ->物理执行图。

StreamGraph:是 根据用 户 通过 Stream API 编 写 的代 码生 成的 最初的 图 。用来表示程序的拓扑结构。

JobGraph:StreamGraph 经过优化后生成了 JobGraph,提交给 JobManager 的数据结构。主要的优化为,将多个符合条件的节点 chain 在一起作为一个节点,这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。

ExecutionGraph : JobManager根 据JobGraph生 成ExecutionGraph 。ExecutionGraph 是 JobGraph 的并行化版本,是调度层最核心的数据结构。

物理执行图 : JobManager 根 据 ExecutionGraph 对 Job 进 行 调 度 后 , 在 各 个TaskManager 上部署 Task 后形成的“图”,并不是一个具体的数据结构。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 10]

### 并行度 ###

Flink 程序的执行具有**并行、 分布式** 的特性。

在执行过程中,一个流(stream)包含一个或多个分区(stream partition),而每一个算子(operator)可以包含一个或多个子任务(operator subtask),这些子任务在不同的线程、不同的物理机或不同的容器中彼此互不依赖地执行。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 11]

\*\*一个特定算子的 子任务(subtask)的个数被称之为其并行度(parallelism)，\*\*所谓并行度就是并行执行任务的个数。

一般情况下,一个 stream 的并行度,可以认为就是其所有算子中最大的并行度。

一个程序中,不同的算子可能具有不同的并行度。

Stream 在算子之间传输数据的形式可以是 **one-to-one(forwarding)的模式** 也可以是 \*\*redistributing 的模式,\*\*具体是哪一种形式,取决于算子的种类。

One-to-one:stream(比如在 source 和 map operator 之间)维护着分区以及元素的顺序。那意味着 map 算子的子任务看到的元素的个数以及顺序跟 source 算子的子任务生产的元素的个数、顺序相同, map、fliter、flatMap 等算子都是 one-to-one 的对应关系。**类似于 spark 中的窄依赖**

Redistributing:stream(map()跟 keyBy/window 之间或者 keyBy/window 跟 sink之间)的分区会发生改变。每一个算子的子任务依据所选择的 transformation 发送数据到不同的目标任务。例如,keyBy() 基于 hashCode 重分区、broadcast 和 rebalance(轮询)会随机重新分区,这些算子都会引起 redistribute 过程,而 redistribute 过程就类似于Spark 中的 shuffle 过程 。**类似于 spark 中的宽依赖**，但是spark的宽依赖等于洗牌，要等到所有的牌齐了才开始洗，而Redistributing相当于发牌，根据发牌规则，牌到了就发。

### 任务链(Operator Chains) ###

Flink 采用了一种称为任务链的优化技术,可以在特定条件下减少本地通信的开销。为了满足任务链的要求,必须将两个或多个算子设为相同的并行度,并通过本地转发(local forward)的方式进行连接

**相同 并行度 的 one to one 操 作** , Flink 这样相连的算子链接在一起形成一个 task,原来的算子成为里面的一部分(subtask)（**并行度相同、并且是 one-to-one 操作,两个条件缺一不可**）。将算子链接成 task 是非常有效的优化:它能减少线程之间的切换和基于缓存区的数据交换,在减少时延的同时提升吞吐量。链接的行为可以在编程 API 中进行指定。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 12]

**可以在**keyAgg那一步执行**disableChaining()**，那么keyAgg就不会和Sink进行任务的合并了，如果之前有的话，也不会与keyAgg进行合并了，也就是两边都会断；或者通过全局配置env.disableOperateChaining()全局的task都不会进行合并了；如果在keyAgg那一步执行了startNewChain()则表示后面的Sink的任务会断开，Sink自己需要重新开启一个任务。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]: /images/20220829/1c3d4d793c16449bb1cbd59c67f81d71.png
[0072d11d48c1473eb6d46c6b5aa7fc21.png]: /images/20220829/75874798c272451db19df9ab72ab3b10.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]: /images/20220829/5c0ef69843874742ac6159d64fdabc1c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 2]: /images/20220829/83a970a773fc4728b193c5cbd0b5e74d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 3]: /images/20220829/b90e0d96f2c14b879c424f32f96376a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 4]: /images/20220829/2ec02b93223c4591aa8fdd23ecf205db.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 5]: /images/20220829/666456dc615a42e890adac75caeae032.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 6]: /images/20220829/44c50fc2271b49ca9eb05f2d75f6d81e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 7]: /images/20220829/b6b45f32b36b497d8830db78923afdc3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 8]: /images/20220829/a30b29e4f2724186847e777ced3d552e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 9]: /images/20220829/be7a45feaa0f4c178a38d6157b718e27.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 10]: /images/20220829/d5456060081d4dc1b6fdaa7f1bbad5e3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 11]: /images/20220829/82c516bced3f4bb0b71530b3153ab5f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 12]: /images/20220829/03abe41795de4a0eae764e7ce4aace04.png