Flink的运行模式解析

大家好，我是栗子鑫，又和大家见面了，今天主要想为大家讲解一下Flink的运行模式，希望对大家有所帮助，正文如下：

—

简介

在笔者的第一篇文章《浅谈Flink》中简单的介绍了Flink的运行的框架，当时主要介绍的作业粒度的运行流程，本篇博客主要介绍 Flink Runtime 的作业执行的核心机制，主要介绍一个作业的提交如何在集群获取资源，然后实现任务。

—

Flink Runtime整体架构

Flink 是可以运行在多种不同的环境中的，例如，它可以通过单进程多线程的方式直接运行，比如在本地单JVM运行，从而提供调试的能力。它也可以运行在 Yarn 或者 K8S 这种资源管理系统上面，也可以在各种云环境中执行。这里借用一张图来介绍Flink的整体架构：

上图可以看出Flink主要提供了DataStream 和 DataSet API两套API，前者为了流式作业后者为批处理作业。而针对不同数据流API Flink 提供了一套统一的分布式作业执行引擎，也就是 Flink Runtime 这层。也是本篇博客介绍的重点。

Flink Runtime 层的主要架构如上图所示，它展示了Flink在集群上的运行方式，主要结构还是Master和Slave结构，其中左侧白色圈中（AppMaster）的部分即是 master，它负责管理整个集群中的资源和作业；而右侧的两个 TaskManager 则是 Slave，负责提供具体的资源并实际执行作业。

AppMaster 部分又包含了三个组件，即 Dispatcher、ResourceManager 和 JobManager。其中，Dispatcher 负责接收用户提供的作业，并且负责为这新提交的作业拉起一个新的 JobManager 组件。ResourceManager 负责资源的管理，在整个 Flink 集群中只有一个 ResourceManager。JobManager 负责管理作业的执行，在一个 Flink 集群中可能有多个作业同时执行，每个作业都有自己的 JobManager 组件。这三个组件都包含在 AppMaster 进程中。

—

运行流程

针对第二节的描述结构，整个的运行流程分为如下几步：

当用户提交作业的时候，提交脚本会首先启动一个 Client 进程负责作业的编译与提交。它首先将用户编写的代码编译为一个 JobGraph，在这个过程，它还会进行一些检查或优化等工作（比如判断哪些 Operator 可以 Chain 到同一个 Task 中）。然后，Client 将产生的 JobGraph 提交到集群中执行。然后根据提交任务模式来决定提交作业的方式（此时有两种情况，一种是类似于 Standalone 这种 Session 模式，AM 会预先启动，此时 Client 直接与 Dispatcher 建立连接并提交作业即可。另一种是 Per-Job 模式，AM 不会预先启动，此时 Client 将首先向资源管理系统（如 Yarn、K8S）申请资源来启动 AM，然后再向 AM 中的 Dispatcher 提交作业）对提交模式笔者会在下一篇博客讲解，如有感兴趣的可以关注一下。
当Dispatcher 接收到作业后，Dispatcher 会首先启动一个 JobManager 组件，然后 JobManager 会向 ResourceManager 申请资源来启动作业中具体的任务。这时根据 Session 和 Per-Job 模式的区别， TaskExecutor 可能已经启动或者尚未启动。对于Session模式，此时 ResourceManager 中已有记录了 TaskExecutor 注册的资源，可以直接选取空闲资源进行分配。而对于Per-Job模式，ResourceManager 也需要首先向外部资源管理系统申请资源来启动 TaskExecutor，然后等待 TaskExecutor 注册相应资源后再继续选择空闲资源进程分配。
ResourceManager 选择到空闲的 Slot 之后，就会通知相应的 TM “将该 Slot 分配分 JobManager XX ”，然后 TaskExecutor 进行相应的记录后，会向 JobManager 进行注册。JobManager 收到 TaskExecutor 注册上来的 Slot 后，就可以实际提交 Task 了。
TaskExecutor 收到 JobManager 提交的 Task 之后，会启动一个新的线程来执行该 Task。Task 启动后就会开始进行预先指定的计算，并通过数据 Shuffle 模块互相交换数据。

—

资源管理和作业调度

作业调度可以看做是对资源和任务进行匹配的过程。作业调度的基础是首先提供对资源的管理，Flink 中的资源是由 TaskExecutor 上的 Slot 来表示的，在 ResourceManager 中，有一个子组件叫做 SlotManager，它维护了当前集群中所有 TaskExecutor 上的 Slot 的信息与状态，如该 Slot 在哪个 TaskExecutor 中，该 Slot 当前是否空闲等。这里借用一张图来表达Flink 中资源管理功能各模块交互关系：

当 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 会从缓存的请求中选择相应的请求并结束相应的请求过程。

除了正常的通信逻辑外，在 ResourceManager 和 TaskExecutor 之间还存在定时的心跳消息来同步 Slot 的状态。在分布式系统中，消息的丢失、错乱不可避免，这些问题会在分布式系统的组件中引入不一致状态，如果没有定时消息，那么组件无法从这些不一致状态中恢复。此外，当组件之间长时间未收到对方的心跳时，就会认为对应的组件已经失效，并进入到 Failover 的流程。

—

总结

本文主要重点介绍了Flink的RunTime详解，同时还介绍了资源管理和作业调度，希望对大家有所帮助。

关注六只栗子，面试不迷路~

作者栗子鑫

编辑一口栗子