Flink 细粒度资源管理新特性解读

Hi，我是王知无，一个大数据领域的原创作者。

ApacheFlink努力为所有现成的应用程序自动导出合理的默认资源需求。对于希望根据特定场景的知识微调资源消耗的用户，Flink提供细粒度资源管理。

一、可能受益于细粒度资源管理的典型场景包括

1、任务具有显著不同的平行性。

2、整个管道所需的资源太多，无法装入单个slot/任务管理器。

3、不同阶段任务所需资源明显不同的批处理作业

二、工作原理

如Flink体系结构中所述，TaskManager中的任务执行资源分为多个slot。slot是Flink运行时中资源调度和资源需求的基本单元。

通过细粒度资源管理，slot请求包含用户可以指定的特定资源配置文件。Flink将尊重用户指定的资源需求，并从TaskManager的可用资源中动态地削减一个完全匹配的slot。如上所示，需要一个具有0.25内核和1GB内存的slot，Flink为其分配slot1。

以前在Flink中，资源需求只包含所需的slot，没有细粒度资源配置文件，即粗粒度资源管理。TaskManager有固定数量的相同slot来满足这些要求。

对于没有指定资源配置文件的资源需求，Flink将自动决定资源配置文件。目前，它的资源配置文件是根据TaskManager的总资源和TaskManager.numberOfTaskSlots计算的，就像在粗粒度资源管理中一样。如上所示，TaskManager的总资源为1核和4 GB内存，任务slot数设置为2，slot2为0.5核和2 GB内存创建，以满足没有指定资源配置文件的需求。

分配slot1和slot2后，TaskManager中剩余0.25个内核和1 GB内存作为可用资源。这些空闲资源可以进一步分区以满足以下资源需求。

有关更多详细信息，请参阅资源分配策略。

三、它如何提高资源效率

在本节中，我们将深入探讨细粒度资源管理如何提高资源效率，这有助于您了解它是否对您的工作有益。

以前，Flink采用了粗粒度资源管理方法，将任务部署到预定义的、通常相同的slot中，而不知道每个slot包含多少资源。对于许多作业，使用粗粒度资源管理并简单地将所有任务放在一个slot共享组中在资源利用率方面已经足够好了。

对于所有任务都具有相同并行性的许多流作业，每个slot将包含整个管道。理想情况下，所有管道都应该使用大致相同的资源，这可以通过调整相同slot的资源轻松实现。

任务的资源消耗随时间而变化。当一个任务的消耗量减少时，额外的资源可以被另一个消耗量增加的任务使用。这就是所谓的调峰填谷效应，减少了所需的总体资源。

但是，在某些情况下，粗粒度资源管理不能很好地工作。

任务可能具有不同的并行性。有时，这种不同的平行性是无法避免的。例如，源/接收器/查找任务的并行性可能受到外部上游/下游系统的分区和IO负载的限制。在这种情况下，任务较少的slot所需的资源将少于具有整个任务管道的slot。

有时，整个管道所需的资源可能太多，无法放入单个slot/TaskManager中。在这种情况下，需要将管道拆分为多个SSG，这些SSG可能并不总是具有相同的资源需求。

对于批处理作业，并非所有任务都可以同时执行。因此，管道的瞬时资源需求随时间而变化。

尝试使用相同的slot执行所有任务可能会导致非最佳资源利用率。相同slot的资源必须能够满足最高的资源需求，这将浪费其他需求。当涉及像GPU这样昂贵的外部资源时，这种浪费会变得更加难以承受。细粒度资源管理利用不同资源的slot来提高此类场景中的资源利用率。

四、资源分配策略

在本节中，我们将讨论Flink运行时中的slot分区机制和资源分配策略，包括Flink运行时如何选择TaskManager来切割slot，以及如何在本机Kubernetes和Thread上分配TaskManager。请注意，资源分配策略在Flink运行时是可插入的，在这里，我们将在细粒度资源管理的第一步中介绍它的默认实现。将来，用户可能会针对不同的场景选择不同的策略。

如“工作原理”一节所述，Flink将从TaskManager中切出一个完全匹配的slot，用于指定资源的slot请求。内部流程如上图所示。TaskManager将使用全部资源启动，但没有预定义的slot。当一个具有0.25内核和1GB内存的slot请求到达时，Flink将选择一个具有足够可用资源的TaskManager，并使用请求的资源创建一个新slot。如果slot被释放，它会将其资源返回到TaskManager的可用资源。

在当前的资源分配策略中，Flink将遍历所有已注册的TaskManager，并选择第一个拥有足够空闲资源以满足slot请求的TaskManager。当没有足够可用资源的TaskManager时，Flink将在本机Kubernetes或Thread上部署时尝试分配新的TaskManager。在当前策略中，Flink将根据用户的配置分配相同的TaskManager。由于TaskManager的资源规格是预定义的：

群集中可能有资源片段。例如，如果有两个slot请求具有3 GB堆内存，而TaskManager的总堆内存为4 GB，则Flink将启动两个TaskManager，每个TaskManager中将浪费1 GB堆内存。将来，可能会有一种资源分配策略，可以根据作业的slot请求分配异构TaskManager，从而减少资源碎片。

您需要确保为slot共享组配置的资源组件不大于TaskManager的总资源。否则，您的工作将以异常的方式失败。

五、局限性

由于细粒度资源管理是一项新的实验性功能，因此并非默认调度器支持的所有功能都可以使用。Flink社区正在努力解决这些限制。

不支持弹性缩放。弹性伸缩目前只支持没有指定资源的slot请求。

不支持任务管理器冗余。slotmanager.redundant-taskmanager-num用于启动冗余taskmanager以加快作业恢复。此配置选项目前不会在细粒度资源管理中生效。

不支持均匀分布的slot策略。此策略尝试在所有可用的TaskManager上均匀分布slot。细粒度资源管理和群集的第一个版本不支持此策略。均匀分布的slot目前不会在其中生效。

与Flink的Web UI的集成有限。细粒度资源管理中的slot可以具有不同的资源规格。web UI目前只显示slot号，而不显示其详细信息。

与批处理作业的有限集成。目前，细粒度资源管理要求在所有边缘类型都被阻塞的情况下执行批处理工作负载。为此，您需要将fine-grained.shuffle-mode.all-blocking配置为true。请注意，这可能会影响性能。有关更多详细信息，请参阅FLINK-20865。

不建议使用混合资源需求。不建议仅为作业的某些部分指定资源需求，而未指定其余部分的需求。目前，任何资源的slot都可以满足未指定的需求。它获取的实际资源可能在不同的作业执行或故障切换中不一致。

slot分配结果可能不是最优的。由于时隙需求包含多个维度的资源，因此时隙分配实际上是一个多维包装问题，是NP-hard问题。默认的资源分配策略可能无法实现最佳的时隙分配，并且在某些情况下可能导致资源碎片或资源分配失败。

配置：cluster.fine-grained-resource-management.enabled

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

SlotSharingGroup ssgA = SlotSharingGroup.newBuilder("a")

.setCpuCores(1.0)

.setTaskHeapMemoryMB(100)

.build();

SlotSharingGroup ssgB = SlotSharingGroup.newBuilder("b")

.setCpuCores(0.5)

.setTaskHeapMemoryMB(100)

.build();

someStream.filter(...).slotSharingGroup("a") // Set the slot sharing group with name “a”.map(...).slotSharingGroup(ssgB); // Directly set the slot sharing group with name and resource.env.registerSlotSharingGroup(ssgA); // Then register the resource of group “a”