Apache Flink 努力为所有应用程序自动推导出合理的默认资源要求。对于希望根据对特定场景的了解来微调资源消耗的用户，Flink提供了细粒度的资源管理。

本页介绍细粒度资源管理的用法、适用场景及其工作原理。

注意：此功能目前是 MVP（“最小可行产品”）功能，仅适用于 DataStream API。

适用场景

可能受益于细粒度资源管理的典型方案包括：

• 任务具有明显不同的并行性。
• 整个管道所需的资源太多，无法放入单个槽/任务管理器中。
• 批处理作业，其中不同阶段的任务所需的资源明显不同

在如何提高资源效率中，深入探讨了细粒度资源管理如何提高上述场景的资源效率。

它是如何运作的

如 Flink 架构中所述，任务管理器中的任务执行资源被拆分为多个槽。该槽是 Flink 运行时中资源调度和资源需求的基本单位。

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

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

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

分配插槽 1 和插槽 2 后，任务管理器中将剩下 0.25 个核心内存和 1 GB 内存作为可用资源。可以对这些可用资源进行进一步分区，以满足以下资源要求。

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

用法

要使用细粒度资源管理，您需要：

• 配置以启用细粒度资源管理。
• 指定资源要求。

实现细粒度资源管理

若要启用细粒度资源管理，需要将群集配置为“细粒度资源管理”。启用 true。

如果没有此配置，Flink 运行时将无法根据您指定的资源要求调度插槽，并且作业将失败并出现异常。

指定插槽共享组的资源要求

细粒度资源要求在插槽共享组上定义。插槽共享组是一个提示，告诉其中的JobManager 算子/任务可以放入同一插槽中。

要指定资源要求，您需要：

• 定义插槽共享组及其包含的算子。
• 指定插槽共享组的资源

有两种方法可以定义插槽共享组及其包含的算子：

• 您可以仅按名称定义插槽共享组，并通过 slot 共享组（字符串名称）将其附加到算子。
• 您可以构造一个 SlotSharingGroup 实例，其中包含插槽共享组的名称和可选资源配置文件。SlotSharingGroup 可以通过 slotSharingGroup（SlotSharingGroup ssg） 附加到算子。

您可以为插槽共享组指定资源配置文件：

• 如果通过 slotSharingGroup（SlotSharingGroup ssg） 设置槽共享组，则可以在构造 SlotSharingGroup 实例时指定资源配置文件。
• 如果仅使用 slot 共享组（字符串名称）设置插槽共享组的名称。您可以构造一个具有相同名称的 SlotSharingGroup 实例以及资源配置文件，并向 StreamExecutionEnvironment#registerSlotSharingGroup（SlotSharingGroup ssg） 注册这些实例的资源。

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”
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

注意：每个插槽共享组只能附加到一个指定的资源，任何冲突都将使您的作业编译失败。

在构造插槽共享组时，可以为插槽共享组设置以下资源组件：

• CPU 内核。定义需要多少个 CPU 内核。需要显式配置正值。
• 任务堆内存。定义需要多少任务堆内存。需要显式配置正值。
• 任务堆外内存。定义需要多少任务堆外内存，可以是 0。
• 托管内存。定义需要多少任务托管内存，可以是 0。
• 外部资源。定义所需的外部资源，可以是空的。

SlotSharingGroup ssgWithResource =
    SlotSharingGroup.newBuilder("ssg")
        .setCpuCores(1.0) // required
        .setTaskHeapMemoryMB(100) // required
        .setTaskOffHeapMemoryMB(50)
        .setManagedMemory(MemorySize.ofMebiBytes(200))
        .setExternalResource("gpu", 1.0)
        .build();

// Build a slot sharing group without specific resource and then register the resource of it in StreamExecutionEnvironment
SlotSharingGroup ssgWithName = SlotSharingGroup.newBuilder("ssg").build();
env.registerSlotSharingGroup(ssgWithResource);
1
2
3
4
5
6
7
8
9
10
11
12

注： 您可以构造一个 SlotSharingGroup，同时指定或不指定其资源配置文件。通过指定资源配置文件，您需要显式设置具有正值的 CPU 内核和任务堆内存，其他组件是可选的。

局限性

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

• 不支持弹性伸缩。弹性伸缩目前仅支持没有指定资源的槽请求。
• 不支持任务管理器冗余。slotmanager.redundant-taskmanager-num 用于启动冗余 TaskManagers 以加快作业恢复。此配置选项目前不会在细粒度资源管理中生效。
• 不支持均匀分布的老虎机策略。此策略尝试在所有可用的任务管理器中均匀分布插槽。该策略在细粒度资源管理和群集的第一个版本中不受支持，均匀分布插槽目前不会在其中生效。
• 与 Flink 的 Web UI 的有限集成。细粒度资源管理中的槽可以具有不同的资源规格。Web UI 目前仅显示插槽号，而不显示其详细信息。
• 与批处理作业的集成有限。目前，细粒度资源管理要求以阻塞所有边缘的类型执行批处理工作负载。为此，您需要将 fine-grained.shuffle-mode.all-blocking 配置为 true。请注意，这可能会影响性能。有关详细信息，请参阅 FLINK-20865。
• 不建议使用混合资源要求。不建议仅指定作业某些部分的资源要求，而未指定其余部分的资源要求。目前，可以使用任何资源的插槽来满足未指定的要求。它获取的实际资源在不同的作业执行或故障转移中可能不一致。
• 插槽分配结果可能不是最佳的。由于槽需求包含多维资源，槽位分配确实是一个多维包装问题，是NP硬的。默认资源分配策略可能无法实现最佳槽分配，并且在某些情况下可能导致资源碎片或资源分配失败。

通知

• 设置插槽共享组可能会改变性能。将可链式算子设置为不同的插槽共享组可能会破坏算子链，从而改变性能。
• 插槽共享组不会限制算子的调度。插槽共享组仅提示调度程序，分组的算子可以部署到共享插槽中。不能保证计划程序始终将分组的算子部署在一起。如果将分组算子部署到单独的插槽中，则插槽资源将从指定的组要求派生。

深入探讨

它如何提高资源效率

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

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

• 对于所有任务都具有相同并行度的许多流式处理作业，每个槽将包含整个管道。理想情况下，所有管道都应使用大致相同的资源，这可以通过调整相同槽的资源轻松满足。
• 任务的资源消耗随时间而变化。当一个任务的消耗量减少时，额外的资源可以由消耗量增加的另一个任务使用。这被称为调峰和谷灌浆效果，减少了所需的整体资源。

但是，在某些情况下，粗粒度资源管理效果不佳。

• 任务可能具有不同的并行性。有时，这种不同的并行性是无法避免的。例如，源/接收器/查找任务的并行性可能受到外部上/下游系统的分区和 IO 负载的限制。在这种情况下，任务较少的槽比具有整个任务管道的槽需要的资源更少。
• 有时，整个管道所需的资源可能太多，无法放入单个插槽/任务管理器中。在这种情况下，管道需要拆分为多个 SSG，这些 SSG 可能并不总是具有相同的资源要求。
• 对于批处理作业，并非所有任务都可以同时执行。因此，管道的瞬时资源需求随时间而变化。

尝试执行具有相同槽位的所有任务可能会导致资源利用率不理想。相同插槽的资源必须能够满足最高的资源要求，这将浪费其他要求。当涉及昂贵的外部资源（如GPU）时，这种浪费可能会变得更加难以承受。细粒度资源管理利用不同资源的槽位来提高此类场景下的资源利用率。

资源分配策略

在本节中，我们将讨论 Flink 运行时中的槽分区机制和资源分配策略，包括 Flink 运行时如何选择 TaskManager 来剪切槽，并在 Native Kubernetes 和 YARN 上分配 TaskManagers。请注意，资源分配策略在 Flink 运行时中是可插入的，在这里，我们将在细粒度资源管理的第一步中介绍其默认实现。将来，用户可能会为不同的方案选择各种策略。

如“工作原理”部分所述，Flink 将从 TaskManager 中为具有指定资源的插槽请求切出一个完全匹配的插槽。内部流程如上所示。任务管理器将使用总资源启动，但没有预定义的插槽。当具有 0.25 Core 和 1GB 内存的插槽请求到达时，Flink 将选择具有足够可用资源的 TaskManager，并使用请求的资源创建一个新插槽。如果释放了某个插槽，它将其资源返回到 TaskManager 的可用资源。

在当前的资源分配策略中，Flink 将遍历所有已注册的任务管理器，并选择第一个具有足够可用资源来满足槽位请求的 TaskManager。当没有足够可用资源的 TaskManager 时，Flink 会在 Native Kubernetes 或 YARN 上部署时尝试分配一个新的 TaskManager。在当前策略中，Flink 会根据用户的配置分配相同的任务管理器。由于 TaskManagers 的资源规范是预定义的：

• 集群中可能存在资源碎片。例如，如果有两个插槽请求具有3 GB堆内存，而TaskManager的总堆内存为4 GB，Flink将启动两个TaskManagers，并且每个TaskManager中将浪费1 GB堆内存。将来，可能会有一种资源分配策略，可以根据作业的槽请求分配异构 TaskManagers，从而缓解资源碎片。
• 您需要确保为插槽共享组配置的资源组件不大于 TaskManager 的总资源。否则，您的作业将失败并出现异常。