Flink 运行时[Runtime] 整体架构
作者:mmseoamin日期:2023-12-25

一、基本组件栈

在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,用户可以直接操作状态和时间等底层数据。

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第1张

【2】Runtime核心层: 该层主要负责对上层不同接口提供基础服务,也是Flink分布式计算框架的核心实现层,支持分布式Stream作业的执行、JobGraph到ExecutionGraph的映射转换、任务调度等。将DataSteam(流作业)和DataSet(批作业)转成统一的可执行的Task Operator,达到在流式引擎下同时处理批量计算和流式计算的目的。其中Runtime层对不同的执行环境提供了一套统一的分布式作业执行引擎。

【3】物理部署层: 该层主要涉及Flink的部署模式,目前Flink支持多种部署模式:本地、集群(Standalone/YARN)、云(GCE/EC2)、Kubenetes。Flink能够通过该层能够支持不同平台的部署,用户可以根据需要选择使用对应的部署模式。

二、Runtime层总体架构

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第2张

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 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第3张

三、资源管理与作业调度

作业调度可以看做是对资源和任务进行匹配的过程。如上所述,在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会从缓存的请求中选择相应的请求并结束相应的请求过程。

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第4张

当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也提供了一种简单的保持负载均衡的方式。

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第5张

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

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第6张

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 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第7张

四、错误恢复

在Flink作业的执行过程中,除正常执行的流程外,还有可能由于环境等原因导致各种类型的错误。整体上来说,错误可能分为两大类:Task执行出现错误或Flink集群的Master出现错误。由于错误不可避免,为了提高可用性,Flink需要提供自动错误恢复机制来进行重试。

Task执行错误:Flink提供了多种不同的错误恢复策略。如下图所示,第一种策略是 Restart-all,即直接重启所有的Task。对于Flink的流任务,由于Flink提供了Checkpoint机制,因此当任务重启后可以直接从上次的Checkpoint 开始继续执行。因此这种方式更适合于流作业。

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第8张

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

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第9张

由于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缓存的输出结果继续进行计算。

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第10张

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

Flink 运行时[Runtime] 整体架构,[点击并拖拽以移动] ​,第11张

如果是由于上游失败导致的错误,那么需要同时重启上游的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进行进一步的优化来避免不必要的作业重启。