Flink1.11内存机制之进程内存及TaskManager内存

问题导读：
1、如何配置 Flink 进程的内存？
2、如何配置总内存？
3、如何配置 TaskManager 内存？
4、如何配置堆内存和托管内存？



Apache Flink 基于 JVM 的高效处理能力，依赖于其对各组件内存用量的细致掌控。考虑到用户在 Flink 上运行的应用的多样性，尽管社区已经努力为所有配置项提供合理的默认值，仍无法满足所有情况下的需求。为了给用户生产提供最大化的价值， Flink 允许用户在整体上以及细粒度上对集群的内存分配进行调整。



本文接下来介绍的内存配置方法适用于 1.10 及以上版本的 TaskManager 进程和 1.11 及以上版本的 JobManager 进程。Flink 在 1.10 和 1.11 版本中对内存配置部分进行了较大幅度的改动，从早期版本升级的用户请参考升级指南。


配置总内存Flink JVM 进程的进程总内存（Total Process Memory）包含了由 Flink 应用使用的内存（Flink 总内存）以及由运行 Flink 的 JVM 使用的内存。Flink 总内存（Total Flink Memory）包括 JVM 堆内存（Heap Memory）和堆外内存（Off-Heap Memory）。其中堆外内存包括直接内存（Direct Memory）和本地内存（Native Memory）。


配置 Flink 进程内存最简单的方法是指定以下两个配置项中的任意一个：

配置项	TaskManager 配置参数	JobManager 配置参数
Flink 总内存	taskmanager.memory.flink.size	jobmanager.memory.flink.size
进程总内存	taskmanager.memory.process.size	jobmanager.memory.process.size

关于本地执行，请分别参考 TaskManager 和 JobManager 的相关文档。

Flink 会根据默认值或其他配置参数自动调整剩余内存部分的大小。关于各内存部分的更多细节，请分别参考 TaskManager 和 JobManager 的相关文档。
对于独立部署模式（Standalone Deployment），如果你希望指定由 Flink 应用本身使用的内存大小，最好选择配置 Flink 总内存。Flink 总内存会进一步划分为 JVM 堆内存和堆外内存。更多详情请参考如何为独立部署模式配置内存。

通过配置进程总内存可以指定由 Flink JVM 进程使用的总内存大小。对于容器化部署模式（Containerized Deployment），这相当于申请的容器（Container）大小，详情请参考如何配置容器内存（Kubernetes、Yarn 或 Mesos）。

此外，还可以通过设置 Flink 总内存的特定内部组成部分的方式来进行内存配置。不同进程需要设置的内存组成部分是不一样的。详情请分别参考 TaskManager 和 JobManager 的相关文档。
提示以上三种方式中，用户需要至少选择其中一种进行配置（本地运行除外），否则 Flink 将无法启动。这意味着，用户需要从以下无默认值的配置参数（或参数组合）中选择一个给出明确的配置：

TaskManager:	JobManager:
taskmanager.memory.flink.size	jobmanager.memory.flink.size
taskmanager.memory.process.size	jobmanager.memory.process.size
taskmanager.memory.task.heap.size 和 taskmanager.memory.managed.size	jobmanager.memory.heap.size

提示不建议同时设置进程总内存和 Flink 总内存。这可能会造成内存配置冲突，从而导致部署失败。额外配置其他内存部分时，同样需要注意可能产生的配置冲突。

JVM 参数Flink 进程启动时，会根据配置的和自动推导出的各内存部分大小，显式地设置以下 JVM 参数：

JVM 参数	TaskManager 取值	JobManager 取值
-Xmx 和 -Xms	框架堆内存 + 任务堆内存	JVM 堆内存
-XX:MaxDirectMemorySize （TaskManager 始终设置，JobManager 见注释）	框架堆外内存 + 任务堆外内存(*) + 网络内存	堆外内存 (*) (**)
-XX:MaxMetaspaceSize	JVM Metaspace	JVM Metaspace

(*) 请注意，堆外内存也包括了用户代码使用的本地内存（非直接内存）。
(**) 只有在 jobmanager.memory.enable-jvm-direct-memory-limit 设置为 true 时，JobManager 才会设置 JVM 直接内存限制。

相关内存部分的配置方法，请同时参考 TaskManager 和 JobManager 的详细内存模型。

受限的等比内存部分本节介绍下列内存部分的配置方法，它们都可以通过指定在总内存中所占比例的方式进行配置，同时受限于相应的的最大/最小值范围。

• JVM 开销：可以配置占用进程总内存的固定比例
• 网络内存：可以配置占用 Flink 总内存的固定比例（仅针对 TaskManager）
  

相关内存部分的配置方法，请同时参考 TaskManager 和 JobManager 的详细内存模型。
这些内存部分的大小必须在相应的最大值、最小值范围内，否则 Flink 将无法启动。最大值、最小值具有默认值，也可以通过相应的配置参数进行设置。例如，如果仅配置下列参数：

• 进程总内存 = 1000Mb
• JVM 开销最小值 = 64Mb
• JVM 开销最大值 = 128Mb
• JVM 开销占比 = 0.1
  

那么 JVM 开销的实际大小将会是 1000Mb x 0.1 = 100Mb，在 64-128Mb 的范围内。
如果将最大值、最小值设置成相同大小，那相当于明确指定了该内存部分的大小。
如果没有明确指定内存部分的大小，Flink 会根据总内存和占比计算出该内存部分的大小。计算得到的内存大小将受限于相应的最大值、最小值范围。例如，如果仅配置下列参数：

• 进程总内存 = 1000Mb
• JVM 开销最小值 = 128Mb
• JVM 开销最大值 = 256Mb
• JVM 开销占比 = 0.1
  

那么 JVM 开销的实际大小将会是 128Mb，因为根据总内存和占比计算得到的内存大小 100Mb 小于最小值。
如果配置了总内存和其他内存部分的大小，那么 Flink 也有可能会忽略给定的占比。这种情况下，受限的等比内存部分的实际大小是总内存减去其他所有内存部分后剩余的部分。这样推导得出的内存大小必须符合最大值、最小值范围，否则 Flink 将无法启动。例如，如果仅配置下列参数：

• 进程总内存 = 1000Mb
• 任务堆内存 = 100Mb（或 JobManager 的 JVM 堆内存）
• JVM 开销最小值 = 64Mb
• JVM 开销最大值 = 256Mb
• JVM 开销占比 = 0.1
  

进程总内存中所有其他内存部分均有默认大小，包括 TaskManager 的托管内存默认占比或 JobManager 的默认堆外内存。因此，JVM 开销的实际大小不是根据占比算出的大小（1000Mb x 0.1 = 100Mb），而是进程总内存中剩余的部分。这个剩余部分的大小必须在 64-256Mb 的范围内，否则将会启动失败。


配置 TaskManager 内存Flink 的 TaskManager 负责执行用户代码。根据实际需求为 TaskManager 配置内存将有助于减少 Flink 的资源占用，增强作业运行的稳定性。

• 配置总内存
• 配置堆内存和托管内存
  
  • 任务（算子）堆内存
  • 托管内存
    
• 配置堆外内存（直接内存或本地内存）
• 内存模型详解
• 框架内存
• 本地执行
  

本文接下来介绍的内存配置方法适用于 1.10 及以上版本。Flink 在 1.10 版本中对内存配置部分进行了较大幅度的改动，从早期版本升级的用户请参考升级指南。
提示本篇内存配置文档仅针对 TaskManager！与 JobManager 相比，TaskManager 具有相似但更加复杂的内存模型。

配置总内存Flink JVM 进程的进程总内存（Total Process Memory）包含了由 Flink 应用使用的内存（Flink 总内存）以及由运行 Flink 的 JVM 使用的内存。其中，Flink 总内存（Total Flink Memory）包括 JVM 堆内存（Heap Memory）、托管内存（Managed Memory）以及其他直接内存（Direct Memory）或本地内存（Native Memory）。



如果你是在本地运行 Flink（例如在 IDE 中）而非创建一个集群，那么本文介绍的配置并非所有都是适用的，详情请参考本地执行。
其他情况下，配置 Flink 内存最简单的方法就是配置总内存。此外，Flink 也支持更细粒度的内存配置方式。
Flink 会根据默认值或其他配置参数自动调整剩余内存部分的大小。接下来的章节将介绍关于各内存部分的更多细节。


配置堆内存和托管内存如配置总内存中所述，另一种配置 Flink 内存的方式是同时设置任务堆内存和托管内存。通过这种方式，用户可以更好地掌控用于 Flink 任务的 JVM 堆内存及 Flink 的托管内存大小。
Flink 会根据默认值或其他配置参数自动调整剩余内存部分的大小。关于各内存部分的更多细节，请参考相关文档。
提示如果已经明确设置了任务堆内存和托管内存，建议不要再设置进程总内存或 Flink 总内存，否则可能会造成内存配置冲突。

任务（算子）堆内存如果希望确保指定大小的 JVM 堆内存给用户代码使用，可以明确指定任务堆内存（taskmanager.memory.task.heap.size）。指定的内存将被包含在总的 JVM 堆空间中，专门用于 Flink 算子及用户代码的执行。

托管内存托管内存是由 Flink 负责分配和管理的本地（堆外）内存。以下场景需要使用托管内存：

• 流处理作业中用于 RocksDB State Backend。
• 批处理作业中用于排序、哈希表及缓存中间结果。
  

可以通过以下两种范式指定托管内存的大小：

• 通过 taskmanager.memory.managed.size 明确指定其大小。
• 通过 taskmanager.memory.managed.fraction 指定在Flink 总内存中的占比。
  

当同时指定二者时，会优先采用指定的大小（Size）。若二者均未指定，会根据默认占比进行计算。
请同时参考如何配置 State Backend 内存以及如何配置批处理作业内存。

配置堆外内存（直接内存或本地内存）用户代码中分配的堆外内存被归为任务堆外内存（Task Off-heap Memory），可以通过 taskmanager.memory.task.off-heap.size 指定。
提示你也可以调整框架堆外内存（Framework Off-heap Memory）。这是一个进阶配置，建议仅在确定 Flink 框架需要更多的内存时调整该配置。
Flink 将框架堆外内存和任务堆外内存都计算在 JVM 的直接内存限制中，请参考 JVM 参数。
提示本地内存（非直接内存）也可以被归在框架堆外内存或任务堆外内存中，在这种情况下 JVM 的直接内存限制可能会高于实际需求。
提示网络内存（Network Memory）同样被计算在 JVM 直接内存中。Flink 会负责管理网络内存，保证其实际用量不会超过配置大小。因此，调整网络内存的大小不会对其他堆外内存有实质上的影响。
请参考内存模型详解。


内存模型详解


如上图所示，下表中列出了 Flink TaskManager 内存模型的所有组成部分，以及影响其大小的相关配置参数。

组成部分	配置参数	描述
框架堆内存（Framework Heap Memory）	taskmanager.memory.framework.heap.size	用于 Flink 框架的 JVM 堆内存（进阶配置）。
任务堆内存（Task Heap Memory）	taskmanager.memory.task.heap.size	用于 Flink 应用的算子及用户代码的 JVM 堆内存。
托管内存（Managed memory）	taskmanager.memory.managed.size taskmanager.memory.managed.fraction	由 Flink 管理的用于排序、哈希表、缓存中间结果及 RocksDB State Backend 的本地内存。
框架堆外内存（Framework Off-heap Memory）	taskmanager.memory.framework.off-heap.size	用于 Flink 框架的堆外内存（直接内存或本地内存）（进阶配置）。
任务堆外内存（Task Off-heap Memory）	taskmanager.memory.task.off-heap.size	用于 Flink 应用的算计及用户代码的堆外内存（直接内存或本地内存）。
网络内存（Network Memory）	taskmanager.memory.network.min taskmanager.memory.network.max taskmanager.memory.network.fraction	用于任务之间数据传输的直接内存（例如网络传输缓冲）。该内存部分为基于 Flink 总内存的受限的等比内存部分。
JVM Metaspace	taskmanager.memory.jvm-metaspace.size	Flink JVM 进程的 Metaspace。
JVM 开销	taskmanager.memory.jvm-overhead.min taskmanager.memory.jvm-overhead.max taskmanager.memory.jvm-overhead.fraction	用于其他 JVM 开销的本地内存，例如栈空间、垃圾回收空间等。该内存部分为基于进程总内存的受限的等比内存部分。

我们可以看到，有些内存部分的大小可以直接通过一个配置参数进行设置，有些则需要根据多个参数进行调整。

框架内存通常情况下，不建议对框架堆内存和框架堆外内存进行调整。除非你非常肯定 Flink 的内部数据结构及操作需要更多的内存。这可能与具体的部署环境及作业结构有关，例如非常高的并发度。此外，Flink 的部分依赖（例如 Hadoop）在某些特定的情况下也可能会需要更多的直接内存或本地内存。
提示不管是堆内存还是堆外内存，Flink 中的框架内存和任务内存之间目前是没有隔离的。对框架和任务内存的区分，主要是为了在后续版本中做进一步优化。

本地执行如果你是将 Flink 作为一个单独的 Java 程序运行在你的电脑本地而非创建一个集群（例如在 IDE 中），那么只有下列配置会生效，其他配置参数则不会起到任何效果：

组成部分	配置参数	本地执行时的默认值
任务堆内存	taskmanager.memory.task.heap.size	无穷大
任务堆外内存	taskmanager.memory.task.off-heap.size	无穷大
托管内存	taskmanager.memory.managed.size	128Mb
网络内存	taskmanager.memory.network.min taskmanager.memory.network.max	64Mb

本地执行模式下，上面列出的所有内存部分均可以但不是必须进行配置。如果未配置，则会采用默认值。其中，任务堆内存和任务堆外内存的默认值无穷大（Long.MAX_VALUE 字节），以及托管内存的默认值 128Mb 均只针对本地执行模式。
提示这种情况下，任务堆内存的大小与实际的堆空间大小无关。该配置参数可能与后续版本中的进一步优化相关。本地执行模式下，JVM 堆空间的实际大小不受 Flink 掌控，而是取决于本地执行进程是如何启动的。如果希望控制 JVM 的堆空间大小，可以在启动进程时明确地指定相关的 JVM 参数，即 -Xmx 和 -Xms。

来源：
https://ci.apache.org/projects/f ... mory/mem_setup.html
https://ci.apache.org/projects/f ... y/mem_setup_tm.html