Flink 运行时[Runtime] 整体架构

一、基本组件栈

在Flink整个软件架构体系中，同样遵循着分层的架构设计理念，在降低系统耦合度的同时，也为上层用户构建Flink应用提供了丰富且友好的接口。从下图中可以看出整个Flink的架构体系基本上可以分为三层，由上往下依次是 API & Libraries层、Runtime核心层以及物理部署层。
【1】API&Libraries层： 作为分布式数据处理框架，Flink同时提供了支撑流计算和批计算的接口，同时在此基础之上抽象出不同的应用类型的组件库，如基于流处理的CEP（复杂事件处理库）、SQL&Table库和基于批处理的FlinkML（机器学习库）等、Gelly（图处理库）等。API层包括构建流计算应用的DataStream API和批计算应用的DataSet API，两者都提供给用户丰富的数据处理高级API，例如Map、FlatMap操作等，同时也提供比较低级的Process Function API，用户可以直接操作状态和时间等底层数据。

【2】Runtime核心层： 该层主要负责对上层不同接口提供基础服务，也是Flink分布式计算框架的核心实现层，支持分布式Stream作业的执行、JobGraph到ExecutionGraph的映射转换、任务调度等。将DataSteam（流作业）和DataSet（批作业）转成统一的可执行的Task Operator，达到在流式引擎下同时处理批量计算和流式计算的目的。其中Runtime层对不同的执行环境提供了一套统一的分布式作业执行引擎。
【3】物理部署层： 该层主要涉及Flink的部署模式，目前Flink支持多种部署模式：本地、集群（Standalone/YARN）、云（GCE/EC2）、Kubenetes。Flink能够通过该层能够支持不同平台的部署，用户可以根据需要选择使用对应的部署模式。

二、Runtime层总体架构

Flink采用了非常经典的Master-Slave结构，Master就对应白线框起来的Dispatcher(负责接收用户提供的作业，并且负责为这个新提交的作业拉起一个新的JobManager组件，在整个Flink集群中只有一个Dispatcher)+ResourceManager(负责资源的管理，在整个Flink集群中只有一个ResourceManager)+JobManager(负责管理作业的执行，在一个Flink集群中可能有多个作业同时执行，每个作业都有自己的JobManager 组件)，这三个组件都包含在AppMaster进程中。Slave就对应TaskManager负责作业的实际执行。
【1】Client： 基于上述结构，当用户提交作业的时候，提交脚本会首先启动一个Client进程负责作业的编译与提交。它首先将用户编写的流式处理代码编译为一个JobGraph，在这个过程，它还会进行一些检查或优化等工作，例如判断哪些Operator可以 Chain到同一个Task中（合并）。然后，Client将产生的JobGraph提交到集群中执行。此时有两种情况，一种是类似于Standalone这种Session模式，AM(Flink Master白框中的组件)会预先启动，此时Client直接与Dispatcher建立连接并提交作业即可。另一种是Per-Job模式，AM不会预先启动，此时Client将首先向资源管理系统 （如Yarn、K8S）申请资源来启动AM，然后再向AM中的 Dispatcher提交作业。
【2】AM： 当作业到Dispatcher后，Dispatcher会首先启动一个JobManager组件，然后JobManager会向ResourceManager申请资源来启动作业中具体的任务。这时根据Session和Per-Job模式的区别， TaskExecutor可能已经启动或者尚未启动。如果是前者，此时ResourceManager中已有记录了TaskExecutor注册的资源，可以直接选取空闲资源进行分配。否则，ResourceManager也需要首先向外部资源管理系统申请资源来启动TaskExecutor，然后等待TaskExecutor注册相应资源后再继续选择空闲资源进程分配。目前Flink中TaskExecutor的资源是通过Slot来描述的，一个Slot一般可以执行一个具体的Task，但在一些情况下也可以执行多个相关联的Task，这部分内容将在下文进行详述。ResourceManager选择到空闲的Slot之后，就会通知相应的TM“将该Slot分配分 JobManager XX，然后TaskExecutor进行相应的记录后，会向JobManager进行注册。JobManager收到TaskExecutor注册上来的Slot后，就可以实际提交Task了。TaskExecutor收到JobManager提交的Task之后，会启动一个新的线程来执行该Task。Task启动后就会开始进行预先指定的计算，并通过数据Shuffle模块互相交换数据。

以上就是Flink Runtime层执行作业的基本流程。可以看出，Flink 支持两种不同的模式，即Per-job模式与Session模式。如下图所示，Per-job模式下整个Flink集群只执行单个作业，即每个作业会独享Dispatcher和ResourceManager组件。此外，Per-job模式下AppMaster 和TaskExecutor都是按需申请的。因此，Per-job模式更适合运行执行时间较长的大作业，这些作业对稳定性要求较高，并且对申请资源的时间不敏感。与之对应，在Session模式下，Flink预先启动AppMaster以及一组TaskExecutor，然后在整个集群的生命周期中会执行多个作业。可以看出，Session模式更适合规模小，执行时间短的作业。

三、资源管理与作业调度

作业调度可以看做是对资源和任务进行匹配的过程。如上所述，在Flink中，资源是通过Slot来表示的，每个Slot可以用来执行不同的Task。而在另一端，任务即Job中实际的Task，它包含了待执行的用户逻辑。调度的主要目的就是为了给Task 找到匹配的Slot。逻辑上来说，每个Slot都应该有一个向量来描述它所能提供的各种资源的量，每个Task也需要相应的说明它所需要的各种资源的量。但是实际上在1.9之前，Flink是不支持细粒度的资源描述的，而是统一的认为每个Slot提供的资源和Task需要的资源都是相同的。从1.9开始，Flink 开始增加对细粒度的资源匹配的支持的实现，但这部分功能目前仍在完善中。

作业调度的基础是首先提供对资源的管理，因此我们首先来看下Flink中资源管理的实现。Flink中的资源是由TaskExecutor上的Slot来表示的。如下图所示，在ResourceManager中，有一个子组件叫做SlotManager，它维护了当前集群中所有TaskExecutor上的 Slot 的信息与状态，如该Slot在哪个TaskExecutor中，该Slot当前是否空闲等。当JobManger来为特定Task申请资源的时候，根据当前是Per-job还是Session模式，ResourceManager可能会去申请资源来启动新的TaskExecutor。当TaskExecutor启动之后，它会通过服务发现找到当前活跃的ResourceManager 并进行注册。在注册信息中，会包含该TaskExecutor中所有Slot的信息。 ResourceManager收到注册信息后，其中的SlotManager就会记录下相应的Slot信息。当JobManager为某个Task来申请资源时，SlotManager就会从当前空闲的Slot中按一定规则选择一个空闲的Slot进行分配。当分配完成后，RM会首先向TaskManager发送RPC要求将选定的Slot 分配给特定的JobManager。TaskManager如果还没有执行过该JobManager的Task的话，它需要首先向相应的JobManager建立连接，然后发送提供 Slot的RPC请求。在JobManager中，所有Task的请求会缓存到SlotPool中。当有Slot被提供之后，SlotPool会从缓存的请求中选择相应的请求并结束相应的请求过程。

当Task结束之后，无论是正常结束还是异常结束，都会通知JobManager相应的结束状态，然后在TaskManager端将Slot标记为已占用但未执行任务的状态。JobManager会首先将相应的Slot缓存到SlotPool中，但不会立即释放。这种方式避免了如果将Slot直接还给ResourceManager，在任务异常结束之后需要重启时，需要立刻重新申请Slot的问题。通过延时释放，Failover的Task可以尽快调度回原来的TaskManager，从而加快Failover的速度。当SlotPool中缓存的Slot超过指定的时间仍未使用时，SlotPool就会发起释放该 Slot的过程。与申请Slot的过程对应，SlotPool会首先通知TaskManager来释放该Slot，然后TaskExecutor通知ResourceManager该Slot已经被释放，从而最终完成释放的逻辑。

除了正常的通信逻辑外，在ResourceManager和TaskExecutor之间还存在定时的心跳消息来同步Slot的状态。在分布式系统中，消息的丢失、错乱不可避免，这些问题会在分布式系统的组件中引入不一致状态，如果没有定时消息，那么组件无法从这些不一致状态中恢复。此外，当组件之间长时间未收到对方的心跳时，就会认为对应的组件已经失效，并进入到Failover的流程。在Slot管理基础上，Flink可以将Task调度到相应的Slot当中。如上所述，Flink尚未完全引入细粒度的资源匹配，默认情况下，每个Slot可以分配给一个Task。但是，这种方式在某些情况下会导致资源利用率不高。如图5所示，假如 A、B、C依次执行计算逻辑，那么给 A、B、C分配分配单独的Slot就会导致资源利用率不高。为了解决这一问题，Flink提供了Share Slot的机制。如图下图所示，基于Share Slot，每个Slot中可以部署来自不同JobVertex的多个任务，但是不能部署来自同一个JobVertex的Task。如图下图所示，每个Slot中最多可以部署同一个A、B或C的Task，但是可以同时部署A、B和C的各一个Task。当单个Task占用资源较少时，Share Slot可以提高资源利用率。 此外，Share Slot也提供了一种简单的保持负载均衡的方式。

基于上述Slot管理和分配的逻辑，JobManager负责维护作业中Task执行的状态。如上所述，Client端会向JobManager提交一个JobGraph，它代表了作业的逻辑结构。JobManager会根据JobGraph按并发展开，从而得到JobManager中关键的ExecutionGraph。ExecutionGraph的结构如下图所示，与JobGraph相比，ExecutionGraph中对于每个Task与中间结果等均创建了对应的对象，从而可以维护这些实体的信息与状态。

Flink中的ExecutionGraph是JobGraph 按并发展开所形成的，它是JobMaster中的核心数据结构

在一个Flink Job中是包含多个Task的，因此另一个关键的问题是在Flink中按什么顺序来调度Task。如下图所示，目前Flink提供了两种基本的调度逻辑，即Eager调度与Lazy From Source。Eager调度如其名所示，它会在作业启动时申请资源将所有的Task调度起来。这种调度算法主要用来调度可能没有终止的流作业。与之对应，Lazy From Source则是从Source开始，按拓扑顺序来进行调度。简单来说，Lazy From Source 会先调度没有上游任务的Source任务，当这些任务执行完成时，它会将输出数据缓存到内存或者写入到磁盘中。然后，对于后续的任务，当它的前驱任务全部执行完成后，Flink就会将这些任务调度起来。这些任务会从读取上游缓存的输出数据进行自己的计算。这一过程继续进行直到所有的任务完成计算。

四、错误恢复

在Flink作业的执行过程中，除正常执行的流程外，还有可能由于环境等原因导致各种类型的错误。整体上来说，错误可能分为两大类：Task执行出现错误或Flink集群的Master出现错误。由于错误不可避免，为了提高可用性，Flink需要提供自动错误恢复机制来进行重试。
Task执行错误：Flink提供了多种不同的错误恢复策略。如下图所示，第一种策略是 Restart-all，即直接重启所有的Task。对于Flink的流任务，由于Flink提供了Checkpoint机制，因此当任务重启后可以直接从上次的Checkpoint 开始继续执行。因此这种方式更适合于流作业。

第二类错误恢复策略是Restart-individual，它只适用于 Task之间没有数据传输的情况。这种情况下，我们可以直接重启出错的任务。

由于Flink的批作业没有Checkpoint机制，因此对于需要数据传输的作业，直接重启所有Task会导致作业从头计算，从而导致一定的性能问题。为了增强对Batch作业，Flink在1.9中引入了一种新的Region-Based 的 Failover策略。在一个Flink的Batch作业中Task之间存在两种数据传输方式，一种是Pipeline类型的方式，这种方式上下游Task之间直接通过网络传输数据，因此需要上下游同时运行；另外一种是Blocking类型，如上节所述，这种方式下，上游的Task会首先将数据进行缓存，因此上下游的Task可以单独执行。基于这两种类型的传输，Flink将ExecutionGraph中使用Pipeline方式传输数据的Task的子图叫做Region，从而将整个 ExecutionGraph划分为多个子图。可以看出，Region内的Task必须同时重启，而不同Region的Task由于在Region边界存在 Blocking的边，因此，可以单独重启下游 Region中的Task。基于这一思路 , 如果某个Region中的某个Task执行出现错误，可以分两种情况进行考虑。如下图所示，如果是由于Task本身的问题发生错误，那么可以只重启该Task所属的Region中的Task，这些 Task重启之后，可以直接拉取上游Region缓存的输出结果继续进行计算。

另一方面，如图如果错误是由于读取上游结果出现问题，如网络连接中断、缓存上游输出数据的TaskExecutor异常退出等，那么还需要重启上游Region来重新产生相应的数据。在这种情况下，如果上游Region输出的数据分发方式不是确定性的（如KeyBy、Broadcast是确定性的分发方式，而Rebalance、Random则不是，因为每次执行会产生不同的分发结果），为保证结果正确性，还需要同时重启上游Region所有的下游Region。

如果是由于上游失败导致的错误，那么需要同时重启上游的Region和下游的Region。实际上，如果下游的输出使用了非确定的数据分割方式，为了保持数据一致性，还需要同时重启所有上游Region和下游Region。

除了Task本身执行的异常外，另一类异常是Flink集群的Master进行发生异常。目前Flink支持启动多个Master作为备份，这些Master可以通过ZK来进行选主，从而保证某一时刻只有一个Master在运行。当前活路的Master发生异常时 , 某个备份的Master 可以接管协调的工作。为了保证Master可以准确维护作业的状态，Flink目前采用了一种最简单的实现方式，即直接重启整个作业。实际上，由于作业本身可能仍在正常运行，因此这种方式存在一定的改进空间。

● 更完善的资源管理： 从1.9开始Flink开始了对细粒度资源匹配的支持。基于细粒度的资源匹配，用户可以为TaskExecutor和 Task设置实际提供和使用的CPU、内存等资源的数量，Flink可以按照资源的使用情况进行调度。这一机制允许用户更大范围的控制作业的调度，从而为进一步提高资源利用率提供了基础。
● 统一的 Stream 与 Batch： Flink目前为流和批分别提供了DataStream和DataSet两套接口，在一些场景下会导致重复实现逻辑的问题。未来Flink会将流和批的接口都统一到DataStream之上。
● 更灵活的调度策略： Flink 从1.9开始引入调度插件的支持，从而允许用户来扩展实现自己的调度逻辑。未来Flink也会提供更高性能的调度策略的实现。
● Master Failover 的优化： 如上节所述，目前Flink在Master Failover时需要重启整个作业，而实际上重启作业并不是必须的逻辑。Flink未来会对Master failover进行进一步的优化来避免不必要的作业重启。