1.Flink 运行时的角色

Flink系统架构中包含了两个角色，分别是JobManager和TaskManager，是一个典型的Master-Slave架构。JobManager相当于是Master，TaskManager相当于是Slave。

Anatomy of a Flink Cluster

Clinet 负责将当前的任务提交给JobManager，提交任务的常用方式：命令提交、web页面提交。当Client提交任务之后，客户端可以断开连接（detached mode），也可以保持连接状态来接收任务的报告（attached mode）。
JobManager和TaskManager启动方式，standalone 集群模式，或者交给资源管理器如YARN或Mesos）进行管理。

1.1 JobManager（JVM进程）作用

• 负责整个集群的资源管理与任务管理
• checkpoint
• 协调对故障恢复

note: 在一个集群中只能由一个正在工作（active）的JobManager，如果HA集群，那么其他JobManager一定是standby状态
主要包含三个不同的组件：ResourceManager、Dispatcher、JobMaster

1.1.1: ResourceManager 资源管理器

• 主要负责管理任务管理器（TaskManager）的插槽（slot），TaskManger插槽是Flink中定义的处理资源单元。
• Flink为不同的环境和资源管理工具提供了不同资源管理器，比如YARN、Mesos、K8s，以及standalone部署。
• 在standalone 部署中，ResourceManager只能分配可用TaskManager的插槽，而不能自行启动新的TaskManager。
• 当JobManager申请插槽资源时，ResourceManager会将有空闲插槽的TaskManager分配给JobManager。如果ResourceManager没有足够的插槽来满足JobManager的请求，它还可以向资源提供平台发起会话，以提供启动TaskManager进程的容器。另外，ResourceManager还负责终止空闲的TaskManager，释放计算资源。

1.1.2： Dispatcher 分发器

• Dispatcher 提供了REST接口去接收flink 作业并且为每一个提交的作业开启一个新的JobMaster
• Dispatcher 启动了一个Flink WebUI提供有关作业执行的信息。
• Dispatcher 在架构中可能并不是必需的，这取决于应用提交运行的方式。

1.1.3： JobMaster

• 一个JobMaster负责管理一个单一的执行 JobGraph。
• Flink群集中可以同时运行多个作业，每个作业都有自己的JobMaster。

1.2 TaskManager（JVM进程）作用

• 负责当前节点上的任务运行及当前节点上的资源管理，TaskManager资源通过TaskSlot进行了划分，每个TaskSlot代表的是一份固定资源。例如，具有三个 slots 的 TaskManager 会将其管理的内存资源分成三等份给每个 slot。 划分资源意味着 subtask 之间不会竞争内存资源，但是也意味着它们只拥有固定的资源。注意这里并没有 CPU 隔离，当前 slots 之间只是划分任务的内存资源
• 负责TaskManager之间的数据交换

note: 至少需要一个TaskManager。TaskManager中资源调度的最小单位是Task Slot。TaskManager中Task Slot的最大数量就是并发处理任务的最大数量，当设置的并行度大于总的Task Slot，程序会报错。请注意，多个operators 算子可以在一个Task Slot中执行

2. 任务提交流程

Flink and YARN interact

1. Client首先检查所请求的资源（ApplicationMaster的内存和vcore）是否可用，Client向HDFS上传Flink的Jar包和配置
2. 之后向Yarn ResourceManager提交任务，请求 APPMaster Container
3. ResourceManager分配Container资源并通知对应的NodeManager启动ApplicationMaster。ApplicationMaster启动后加载Flink的Jar包和配置构建环境，然后启动JobManager
4. 之后ApplicationMaster向ResourceManager申请资源启动TaskManager，ResourceManager分配Container资源后，由ApplicationMaster通知资源所在节点的NodeManager启动TaskManager，NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager，TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。

3. Distributed Runtime Environment

3.1 Tasks Parallelism and Operator Chains 并行度和任务链

3.1.1 Tasks Parallelism

• Flink程序的执行具有并行、分布式的特性。
• 在执行过程中，一个流（stream）包含一个或多个分区（stream partition），而每一个算子（operator）可以包含一个或多个子任务（operator subtask），这些子任务在不同的线程、不同的物理机或不同的容器中彼此互不依赖地执行。
• 一个特定算子的子任务（subtask）的个数被称之为其并行度（parallelism）。一般情况下，一个流程序的并行度，可以认为就是其所有算子中最大的并行度。一个程序中，不同的算子可能具有不同的并行度。

image.png

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中的宽依赖

3.1.2 Operator Chains、

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

image.png

3.2 TaskManger、Slots与 source

• 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的受管理的内存。

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

  
  
  before slot sharing,
  
  

  默认情况下，Flink允许子任务共享插槽，即使它们是不同任务的子任务也是如此，只要它们来自同一作业即可。结果是一个插槽可以容纳整个作业管道。允许此插槽共享有两个主要好处：
  1） Flink集群所需的任务槽与作业中使用的最高并行度恰好一样。无需计算一个程序总共包含多少个任务（并行度各不相同）
  2） 更容易获得更好的资源利用率。通过slot sharing，我们示例中的基本并行度从2增加到6，可以充分利用插槽资源，同时确保沉重的子任务在TaskManager之间公平分配。

slot sharing

举个例子：
Task Slot是静态的概念，是指TaskManager具有的并发执行能力，可以通过参数taskmanager.numberOfTaskSlots进行配置；而并行度parallelism是动态概念，即TaskManager运行程序时实际使用的并发能力，可以通过参数parallelism.default进行配置。
也就是说，假设一共有3个TaskManager，每一个TaskManager中的分配3个TaskSlot，也就是每个TaskManager可以接收3个task，一共9个TaskSlot，如果我们设置parallelism.default=1，即运行程序默认的并行度为1，9个TaskSlot只用了1个，有8个空闲，因此，设置合适的并行度才能提高效率。

并行度优化前

并行度优化后

4.执行图（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 后形成的“图”，并不是一个具体的数据结构。
  
  ExecutionGraph

参考 flink 官网 Flink Architecture
flink 官网 Clusters & Deployment YARN
flink 官网 release-1.10 runtime