Flink 资源管理

文章目录

前言
ResourceManager详解
Slot 管理器
SlotProvider
Slot资源池
Slot共享
- Slot共享的优点
- Slot 共享组与 Slot 共享管理器
- Slot资源申请
总结

前言

在Flink中，资源管理是一个核心组件，它负责分配和管理计算资源，以确保任务能够高效、稳定地运行。以下是关于Flink资源管理的详细解释：

资源管理的目标：
- 高效性：确保任务能够充分利用可用的计算资源，达到最佳的处理性能。
- 稳定性：在资源不足或任务失败时，能够优雅地处理并恢复任务，保持系统的稳定运行。
资源管理的组件：
- Flink Cluster：由Flink Master（也称为JobManager）和多个Task Managers组成。Master负责整个集群的协调和资源分配，而Task Managers负责执行具体的任务。
- ResourceManager：位于Flink Master中，负责整个集群的资源调度以及与外部调度系统（如YARN、Kubernetes等）的对接。
- JobManager：每个JobManager负责管理一个具体的作业（Job）。它包含一个Scheduler组件，负责调度执行该Job的DAG（有向无环图）中所有Task，并发出资源请求。
- TaskManager：负责执行具体的Task，并持有Slot资源。Slot是Flink中最小的资源分配单位，一个TaskManager可以有多个Slot。
资源抽象：
- 集群资源：由资源管理框架（如YARN、Kubernetes等）管理，包括CPU、内存、GPU等硬件资源。Flink从资源管理框架中申请和释放这些资源。
- Flink自身资源：Flink对申请到的资源进行切分，每一份称为TaskSlot。Slot是Flink内部对资源的抽象，用于隔离不同作业的任务。
资源申请与释放：
- Flink从资源管理框架申请资源容器（如YARN中的Container或Kubernetes中的Pod），每个容器中运行一个TaskManager进程。
- TaskManager中的每个Slot可以运行一个Task。因为计算类型的不同，一个Task可能无法充分利用一个Slot的资源，所以单个Slot可以被多个Flink的任务共享。
资源调度策略：
- Flink提供了多种资源调度策略，如基于优先级的调度、公平调度等，以满足不同作业的资源需求。
容错与恢复：
- 当TaskManager或JobManager失败时，Flink能够自动检测并触发相应的恢复机制，确保作业的持续运行。
与外部系统的集成：
- Flink可以与多种资源管理框架（如YARN、Kubernetes、Mesos等）集成，实现资源的动态申请与释放。

综上所述，Flink的资源管理是一个复杂而关键的过程，它涉及到集群资源的申请与释放、资源的抽象与切分、资源调度策略以及容错与恢复等多个方面。通过有效的资源管理，Flink能够确保作业的高效、稳定运行。

ResourceManager详解

ResourceManager 作为统一的集群资源管理器，用于管理整个集群的计算资源，包括 CPU资源、内存资源等。同时ResourceManager 负责向集群资源管理器申请容器资源启动TaskManager实例，并对TaskManager进行集中管理。当新的 Job 提交到集群后，JobManager 会向 ResourceManager 申请作业执行需要的计算资源，进而完成整个作业的运行。
在 FlinK 中内置了 4 种 ResourceManager，即 YarnResourceManager、KubernetesResourceManager(K8s)、StandaloneResourceManager、 MesosResourceManager，分别对应于不同的资源管理框架。

Slot 管理器

Slot 管理器在 Flink 中叫作 SlotManager，是 ResourceManager 的组件，从全局角度维护当前有多少个 TaskManager，每个 TaskManager 有多少个空闲的 Slot 和 Slot 等资源的使用情况。当 Flink 作业调度执行时，根据 Slot 分配策略为 Task 分配执行位置。

SlotManager 虽然是 ResourceManager 的组件，但是其逻辑是通用的，并不关心到底使用了哪种资源集群。面向不同的对象，SlotManager 提供不同的功能：

1）对 TaskManager 提供注册、取消注册、空闲退出等管理动作，注册则集群可用的 Slot 变多，取消注册、空闲推出则释放资源，还给资源管理集群。
2）对 Flink 作业，接受 Slot 的请求和释放、资源汇报等。当资源不足的时候，SlotManager 将资源请求暂存在等待队列中，SlotManager 通知 ResourceManager 去申请更多的资源，启动新的 TaskManager，TaskManager 注册到 SlotManager 之后，SlotManager 就有可用的新资源了，从等待队列依次分配资源。

SlotProvider

SlotProvider接口定义了Slot的请求行为，支持两种请求模式。

立即响应模式：Slot请求会立即执行。
排队模式：排队等待可用的Slot，当资源可用时分配资源。

最终的实现在SchedulerImpl中，其中Scheduler接口增加了SlotSelectionStratety。

Slot资源池

Slot资源池在Flink中叫做SlotPool，是JobMaster中记录当前作业从TaskManager获取的Slot的集合。JobMaster的调度器首先从SlotPool中获取Slot来调度任务，SlotPool在没有足够的Slot资源执行作业的时候，首先尝试从ResourceManager中获取资源，如果ResourceManager当前不可用，ResourceManager拒绝资源请求或者请求超时，资源申请失败，则资源启动失败。

JobMaster申请到资源后，会在本地持有Slot，避免ResourceManager异常导致作业运行失败。对于批处理而言，持有资源JobMaster首先可以避免多次向JobMaster申请资源，同时ResourceManager不可用也不会影响作业的正常执行，只有资源不足时，才会导致作业执行失败。

当作业执行完毕或者作业完全启动且资源有剩余时，JobMaster会将剩余的资源交给ResourceManager。

Slot共享

每个TaskManager都是一个Java进程，TaskManager为Task分配独立的执行线程，一个TaskManager中可能执行一个或多个Task。TaskManager通过Slot来控制（一个TaskManager至少有一个Slot）TaskManager能接收多少个Task。

Slot表示TaskManager拥有资源的一个固定大小的子集。假如一个TaskManager有3个Slot，那么它会将其管理的内存分成3份给Slot，在没有Slot共享的情况下，并行度为2的作业部署之后，Slot与Task的分配关系如下图所示。Slot的资源化意味着一个作业的Task将不需要跟来自其他作业的Task竞争内存、CPU等计算资源。
通过调整Slot数量，用户可以定义Task之间如何隔离。如果一个TaskManager只有一个Slot，意味着每个Task独立地运行在JVM进程中。而一个TaskManager有多个Slot，则意味着更多的Task可以共享一个JVM。在同一个JVM进程中的Task将共享TCP连接和心跳消息。Task之间也可能会共享数据集和数据结构，这样可减少每个Task的负载。

虽然通过Slot对TaskManager的资源进行划分，在一定程度上能都提高集群资源的利用率，但这种做法并没有考虑到不同Task的计算任务对资源需求的差异，有时候还会是多种资源混合类型。

所以在Slot的基础上，Flink设计了Slot共享机制。其中，SlotSharingManager用在Flink作业的执行调度中，负责Slot的共享，不同的Task可以共享Slot。

Slot共享的优点

默认情况下，Flink作业共享同一个SlotSharingGroup，同一个作业中来自不同的JobVertex的Task可以共享作业。使用Slot共享，可以在一个Slot中运行Task组成的流水线。共享Slot带来如下优点。

资源分配简单
Flink集群需要的Slot数量和作业中的最高并行度一致，不需要计算一个程序总共包含多少个Task。
资源利用率高
如果没有Slot共享，资源密集型的Task（如长周期的窗口计算）跟非密集型的作业（如Source/Map）占用相同的资源，在整个TaskManager层面上，资源没有充分利用。如果共享Slot，将并行度从2提高至6，可以充分利用Slot资源，同时确保资源密集型的Task在TaskManager中公平分配。

Slot 共享组与 Slot 共享管理器

Slot共享管理器在Flink中叫作SlotSharingManager，Slot共享组在Flink中叫作SlotSharingGroup。SlotSharingManager对象管理资源共享与分配，1个Slot共享组对应1个Slot共享管理器。两者在作业调度执行的时候发挥作用，部署Task之前，选择Slot确定Task发布到哪个TaskManager。

Flink有两种共享组：

SlotSharingGroup
非强制性共享约束，Slot共享根据组内的JobVertices ID查找是否已有可以共享的Slot，只要确保相同的JobVertex ID不能出现在一个共享的Slot内即可。
在符合资源要求的Slot中，找到没有相同JobVertex ID的Slot，根据Slot选择策略选择一个Slot即可，如果没有符合条件的Slot，则申请新的Slot。
CoLocationGroup
CoLocationGroup又叫作本地约束共享组，具有强制性的Slot共享限制，CoLocationGroup用在迭代运算中，即在IterativeStream的API中调用。迭代运算中的Task必须共享同一个TaskManager的Slot。CoLocationGroup可以看成是SlotSharingGroup的特例。
此处需要注意，JobGraph向ExecutionGraph的转换过程中，为每一个ExecutionGraph Vertex赋予了按照并行度编写的编号，相同编号的迭代计算Execuetion Vertex会被放入本地共享约束组中，共享相同的CoLocationConstraint对象，在调度的时候，根据编号就能找到其他Task的Slot信息。
CoLocation共享根据组内每个ExecutionVcrtex关联的CoLocationConsmt查找是否有相CoLocaLionConstraint约束已分配到Slot可用，在调度作业执行的时候，首先要找到本约束中其他Task部署的TaskManager，如果没有则申请一个新的Slot，如果有则共享该TaskManager上的Slot。

Slot资源申请

单独Slot资源申请
该类的Slot申请首先会从JobMaster的当前SlotPool中尝试获取资源，如果资源不足，则从SlotPool中申请新的Slot，然后SlotPool向ResourceManager请求新的Slot。
共享Slot资源申请
共享Slot在申请的时候，需要向SlotSharingManager请求资源，如果有CoLocation限制，则申请CoLocation MultiTaskSlot，否则申请一般的MultiTaskSlot。

SingleTaskSlot表示运行单个Task的Slot，每个SingleTaskSlot对应于一个LogicalSlot。MultiTaskSlot中包含一组TaskSlot。

借助SingleTaskSlot和MultiTaskSlot，Flink实现了一般Slot共享和CoLocaitonGroup共享，两者的数据结构如下图所示：