Flink 核心概念与架构总结

一. 概要介绍

Flink 是一个分布式系统，需要有效分配和管理计算资源才能执行流应用程序。它集成了所有常见的集群资源管理器，例如Hadoop YARN、Apache Mesos和Kubernetes，但也可以设置作为独立集群甚至库运行。

flink on yarn 有两种提交方式

• yarn-session：yarn-session模式需要先启动集群(jobmanager、TaskManager等)，然后再提交作业，接着会向yarn申请一块空间后，资源永远保持不变。如果资源满了，下一个作业就无法提交，只能等到yarn中的其中一个作业执行完成后，释放了资源，下个作业才会正常提交。需要提前运行 yarn-session.sh命令
• yarn-per-job：一个Job会对应一个集群(jobmanager、TaskManager等)，每提交一个作业会根据自身的情况，都会单独向yarn申请资源，直到作业执行完成，一个作业的失败与否并不会影响下一个作业的正常提交和运行。每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成之后创建的集群也会消失。

二. Flink组件

2.1 client

客户端不是运行时和程序执行的一部分，但它用于准备并发送JobGraph给JobManager，然后客户端断开连接或者维持连接以等待接收计算结果。可以是运行在任何机器上（与 JobManager 环境连通即可）。

2.2 JobManager

主要负责调度 Job 并协调 Task 做 checkpoint，从 Client 处接收到 Job 和 JAR 包等资源后，会生成优化后的执行计划，并以 Task 为单元调度到各个 TaskManager 去执行，始终至少有一个JobManager。HA设置中可能有多个JobManager，一个是活跃的。

JobManager包括3个组件：Dispatcher、ResourceManager、JobMaster

• ResourceManager：负责Flink集群中的资源提供、回收、分配。它管理task slots，这是Flink集群中资源调度的最小单位。Flink为不同的环境和资源管理者实现了对应的ResourceManager
• Dispatcher：Dispatcher提供了一个REST接口，用来提交Flink应用程序执行。并为每个提交的作业启动一个JobMatser。它还运行 FlinkWbeUI提供作业执行信息。
• JobMaster：JobMaster负责管理单个JobGrouph的执行。Flink集群中可以同时运行多个作业，每个作业都有自己的JobMaster。

2.3 TaskManager

在启动的时候就设置好了槽位数（Slot），每个 slot 能启动一个 Task，Task 为线程。从 JobManager 处接收需要部署的 Task，部署启动后，与自己的上游建立 Netty 连接，接收数据并处理。每一个TaskManager是一个JVM进程，它可能会在独立的线程上执行一个或多个subtask。

2.4 Slot

每个task slot表示TaskManager拥有资源的一个固定大小的子集。假如一个TaskManager有三个slot，那么它会将其管理的内存分成三份给各个slot。资源slot化意味着一个task将不需要跟来自其他job的task竞争被管理的内存，需要注意的是，这里不会涉及到CPU的隔离，slot目前仅仅用来隔离task的受管理的内存。

如果一个TaskManager一个slot，那将意味着每个subtask运行在独立的JVM中，而一个TaskManager多个slot意味着更多的subtask可以共享同一个JVM。在同一个JVM进程中的task将共享TCP连接（基于多路复用）和心跳消息。也可能共享数据集和数据结构，因此这减少了每个subtask的负载。

2.5 Slot与Parallelism

Task Slot是静态的概念，是指TaskManager具有的并发执行能力，可以通过 taskmanager.numberOfTaskSlots 进行配置，而并行度parallelism是动态概念，即TaskManager运行程序时实际使用的并发能力，可以通过参数parallelism.default进行配置。一个Operator被分成subtask的数量就是并行度（parallelism），它决定程序并发执行的线程个数。一个程序中，不同的operator可能具有不同的并行度。

2.6 Task与SubTask

• 一个算子叫做一个Task，一个算子的并行度是几，这个Task就有几个SubTask。
• 相同并行度的one to one流操作，Flink将这样相连的算子链接在一起形成一个task，一个算子的并行度是几，这个Task就有几个SubTask。

2.7 operator chains

相同并行度的one to one操作，Flink将这样相连的operator 链接在一起形成一个task，将operators链接成task是非常有效的优化：它能减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。

OperatorChain的优点：

• 减少线程切换
• 减少序列化与反序列化
• 减少延迟并且提高吞吐能力

OperatorChain 算子链组成条件：

• 上下游算子并行度一致
• 上下游算子之间没有数据shuffle
• 使用相同的算子共享组，默认情况下使用的是名叫 default 的共享组

三. Flink任务yarn提交流程

Flink on Yanr 任务提交流程，yarn-cluster提交流程per-job模式

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，NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager，TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。

四. Flink执行图分类

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

• StreamGraph： 是根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序的拓扑结构。
• JobGraph： StreamGraph经过优化后生成了 JobGraph，提交给 JobManager 的数据结构。主要的优化为，将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。
• ExecutionGraph： JobManager 根据 JobGraph 生成ExecutionGraph。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。
• 物理执行图： JobManager 根据 ExecutionGraph 对 Job 进行调度后，在各个TaskManager 上部署 Task 后形成的graph，并不是一个具体的数据结构。

五. Flink数据流形式

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

• One-to-one：stream(比如在source和map operator之间)维护着分区以及元素的顺序。那意味着map operator的subtask看到的元素的个数以及顺序跟source operator的subtask生产的元素的个数、顺序相同，map、fliter、flatMap等算子都是one-to-one的对应关系。类似于spark中的窄依赖
• Redistributing：stream的分区会发生改变，比如map()跟keyBy/window之间或者keyBy/window跟sink之间。每一个operator subtask依据所选择的transformation发送数据到不同的目标subtask。例如，keyBy() 基于hashCode重分区、broadcast和rebalance会随机重新分区，这些算子都会引起redistribute过程，而redistribute过程就类似于Spark中的shuffle过程。类似于spark中的宽依赖