flink 内存配置（二）：设置TaskManager内存

TaskManager在Flink中运行用户代码。根据需要配置内存使用，可以极大地减少Flink的资源占用，提高作业的稳定性。
注意下面的讲解适用于TaskManager 1.10之后的版本。与JobManager进程的内存模型相比，TaskManager内存组件具有类似但更复杂的结构。

1. 配置 Total Memory

Flink JVM 进程的总内存（Total Process Memory）由 Flink 应用程序（Total Flink Memory，Flink 总内存）和 JVM 运行进程所消耗的内存组成。Flink 总内存（ Total Flink Memory）包括 JVM Heap、Managed Memory（托管内存，由 Flink 管理）和其他 Direct Memory（或 Native Memory）的使用量。如下结构：

如果在本地运行Flink（例如在IDE中），而没有创建集群，那么只有一部分内存配置选项是相关的，详情见下面的本地执行章节。
否则，为TaskManagers 设置内存的最简单方法是配置总内存。下面将更详细地描述一种更细粒度的方法。
其余的内存组件将根据默认值或额外配置的选项自动调整。有关其他内存组件的更多细节，请参阅后面的章节。

2. 配置 Heap and Managed Memory

前面提到过，在Flink中设置内存的另一种方法是显式指定Task Heap Memory 和 Managed Memory。它让Flink能够更好地控制可用的JVM Heap 和 Managed Memory。

其余的内存组件将根据默认值或额外配置的选项自动调整。下面是关于其他内存组件的更多细节。

如果已明确配置了任务 Task Heap Memory 和 Managed Memory，建议既不要设置进程总内存（Total Process Memory），也不要设置 Flink 总内存（Total Flink Memory）。否则，很容易导致内存配置冲突。

2.1. Task (Operator) Heap Memory

如果你想保证用户代码有一定量的JVM堆可用，可以显式设置Task Heap Memory（taskmanager.memory.task.heap.size）。它会被添加到JVM的堆大小中，并专门用于运行用户代码的 Flink Operator。

2.2. Managed Memory

Managed memory（托管内存）由Flink管理，并作为本地内存（堆外off-heap）分配。以下工作负载使用托管内存：

流式作业可以将其用于RocksDB状态后端。
流和批处理作业都可以使用它来排序、散列表和缓存中间结果。
流和批处理作业都可以使用它在Python进程中执行用户定义的函数。

托管内存大小可以由以下配置：

可通过 taskmanager.memory.managed.size 进行显式配置。
或通过 taskmanager.memory.managed.fraction乘以Flink 总内存（Total Flink Memory）来计算出来。

如果同时设置了大小和比例，则大小优先于比例。如果既没有明确配置大小，也没有配置比例，则将使用默认比例。

如果你的作业包含多种类型的托管内存使用者，你可以控制如何在这些类型之间共享托管内存。配置选项 taskmanager.memory.managed.consumer-weights 可以为每种类型设置权重，Flink会按比例分配托管内存。有效的消费者类型有：

OPERATOR：用于内置算法。
STATE_BACKEND：用于流式传输中的 RocksDB 状态后端。
PYTHON：用于PYTHON进程。

例如，如果一个流作业同时使用RocksDB状态后端和Python udf，并且消费者权重配置为STATE_BACKEND:70， Python:30， Flink将为RocksDB状态后端保留70%的总管理内存，为Python进程保留30%。

对于每种类型，Flink只在作业中包含该类型的托管内存消费者时才分配托管内存。例如，如果一个流作业使用堆状态后端和Python udf，并且消费者权重配置为STATE_BACKEND:70， Python:30， Flink将为Python进程使用其所有的托管内存，因为堆状态后端不使用托管内存。

注意：对于未包含在consumer权重中的consumer类型，Flink不会为其分配托管内存。如果作业确实需要缺失的类型，则可能导致内存分配失败。默认情况下，包含所有消费者类型。只有当权重被显式配置/覆盖时，才会发生这种情况。

3. 配置 Off-Heap Memory（Direct or Native）

由用户代码分配的堆外内存应计入任务堆外内存（taskmanager.memory.task.off-heap.size）。

你也可以调整框架的堆外内存（Framework Off-Heap Memory）。但是只有在确定 Flink 框架需要更多内存时，才应该去更改该值。

Flink 将框架堆外内存（Framework Off-Heap Memory）和任务堆外内存（Task Off-Heap Memory）纳入 JVM 的直接内存（Direct Memory）限制。

注意：虽然本地非直接内存（Native Non-Direct Memory）的使用可以作为框架堆外内存或任务堆外内存的一部分，但在这种情况下会导致更高的 JVM 直接内存限制。

注意：网络内存（Network Memory）也是 JVM 直接内存的一部分，但它由 Flink 管理，并保证永远不会超过其配置大小。因此，调整网络内存大小对这种情况没有帮助。
详见下方的内存详细模型。

4. 详细内存模型

下表列出了如上所述的所有内存组件，以及影响各个组件大小的Flink配置选项。

组件	配置项	描述
Framework Heap Memory	taskmanager.memory.framework.heap.size	专用于Flink框架的JVM堆内存
Task Heap Memory	taskmanager.memory.task.heap.size	JVM堆内存专用于Flink应用程序，用于运行flink operators和用户代码
Managed memory	taskmanager.memory.managed.size taskmanager.memory.managed.fraction	由Flink管理的本地内存，保留被用于排序，哈希表，缓存中间结果和RocksDB状态后端
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	为 Task 之间的数据交换预留的直接内存（例如，通过网络传输的缓冲）是Flink总内存（Total Flink Size）的一个fractionated部分。这些内存用于分配网络缓冲区
JVM metaspace	taskmanager.memory.jvm-metaspace.size	Flink JVM进程的元空间大小
JVM Overhead	taskmanager.memory.jvm-overhead.min taskmanager.memory.jvm-overhead.max taskmanager.memory.jvm-overhead.fraction	本地内存是为其他JVM开销保留的：例如线程堆栈、代码缓存、垃圾收集空间等，它是总进程内存的一个受限制的分块组件

如你所见，一些内存组件的大小可以通过相应的选项简单地设置。其他组件可以使用多个选项进行调优。

5. Framework Memory

如果没有充分的理由，不应该更改框架的堆内存（Framework Heap Memory）和框架的堆外内存（Framework Off-Heap Memory）。只有在确定Flink需要为某些内部数据结构或操作增加内存时，才需要调整内存大小。它可能与特定的部署环境或作业结构相关，如高度并行性。此外，在某些情况下，Flink的依赖（如Hadoop）可能会消耗更多的直接内存或本地内存。

注意：Flink 目前既没有隔离 Framework Heap Memory 和 Task Heap Memory，也没有隔离 Framework Off-Heap Memory 和 Task Off-Heap Memory。框架和任务内存的分离可以在未来的版本中用于进一步的优化。

6. 本地执行（Local Execution）

如果你在本地机器上启动Flink作为单个java程序而不创建集群（例如从你的IDE），那么所有组件都被忽略，除了以下组件：

内存组件	相关选项	默认值
Task heap	taskmanager.memory.task.heap.size	无限制
Task off-heap	taskmanager.memory.task.off-heap.size	无限制
Managed memory	taskmanager.memory.managed.size	128MB
Network memory	taskmanager.memory.network.min taskmanager.memory.network.max	64MB

上面列出的所有组件都可以但不必为本地执行显式配置。如果没有配置，则将其设置为默认值。任务堆内存（Task Heap）和任务堆外内存（Task Off-Heap）被认为是无限大的（（Long.MAX_VALUE 字节）。托管内存（Managed Memory）在本地执行模式下的默认值为128MB。

注意，在这种情况下，任务堆大小与实际堆大小没有任何关系。它可以与下一个版本带来的未来优化相关。已启动的本地进程的实际JVM堆大小并不受Flink控制，而是取决于如何启动进程。如果您想控制JVM堆大小，则必须显式传递相应的JVM参数，例如-Xmx、-Xms。

7. 实际配置效果展示

注意和上图的详细内存模型组成和各个配置项在不配置的时候的默认值结合着分析。即：

Total Process Size = Total Flink Size + JVM Metaspace + JVM Overhead
Total Flink Size = Total Heap Size + Total Off-Heap Size
Total JVM Heap Size = Task Heap + Framework Heap
Total Off-Heap = Managed Memory + Direct Memory（Task Off-Heap + Framework Off-Heap + Network Memory）
JVM Metaspace = taskmanager.memory.jvm-metaspace.size（默认值256m）
JVM Overhead = Total Process Size * taskmanager.memory.jvm-overhead.fraction（默认值0.1）
Task Heap = taskmanager.memory.task.heap.size
Framework Heap = taskmanager.memory.framework.heap.size（默认值128m）
Managed Memory = taskmanager.memory.managed.size（没配置的话就是 Total Flink Size * taskmanager.memory.managed.fraction（默认值0.4））
Task Off-Heap = taskmanager.memory.task.off-heap.size（默认值0）
Framework Off-Heap = taskmanager.memory.framework.off-heap.size（默认值128m）
Network Memory = Total Flink Size * taskmanager.memory.network.fraction（默认值0.1）
其他配置值都可以在官网配置页面查看，注意有些 fraction会有对应的最大最小值限制范围。

情况1：只配置 Total Process Size

这里配置 Total Process Size 为 2048m，即 -Dtaskmanager.memory.process.size=2048m

（这里选择使用application mode 运行在yarn上）

可以看下内存是如何计算出以上配置值的：（下面值乘以系数的做了四舍五入处理，为了和图对应）

首先只配置了 Total Process Size = 2048m，其他就都是取默认值或推导计算出。
JVM Metaspace = 256m
JVM Overhead = 2048m * 0.1 = 205m
Total Flink Size = 2048m - 256m - 205m = 1587m
Task Heap没指定所以先计算其他的值，最终减去其他值得到
Framework Heap = 128m
Managed Memory没有配置taskmanager.memory.managed.size，所以取值：1587m * 0.4 = 635m
Framework Off-Heap = 128m
Task Off-Heap = 0
Network Memory = 1587m * 0.1 = 159m（而且没超过最大最小值限制）
Direct Memory = 0 + 128m + 159m = 287m
Task Heap = Total Flink Size - Framework Heap - Framework Off-Heap - Task Off-Heap - Managed Memory - Network Memory，即 1587m - 128m - 128m - 0 - 635m - 159m = 538m
Total JVM Heap Size = 538m + 128m = 666m

从jobmanager日志和taskmanager日志上也能看到内存分配值：

情况2：只配置 Total Flink Size

这里配置 Total Flink Size 为 2048m，即 -Dtaskmanager.memory.flink.size=2048m

（这里选择使用Session Mode 运行在standalone上，注意使用哪种部署方式无关紧要，这里运行在session模式下，只是看到内存分配日志打印的更为清晰）

测试环境的准备可以使用配置 conf/flink-conf.yaml里配置 taskmanager.memory.flink.size: 2048m

然后执行 bash bin/start-cluster.sh 然后看taskmanager的信息

可以看下内存是如何计算出以上配置值的：（下面值乘以系数的做了四舍五入处理，为了和图对应）

首先只配置了 Total Flink Size = 2048m，其他就都是取默认值或推导计算出。
Task Heap没指定所以先计算其他的值，最终减去其他值得到
Framework Heap = 128m
Managed Memory没有配置taskmanager.memory.managed.size，所以取值：2048m * 0.4 = 819m
Framework Off-Heap = 128m
Task Off-Heap = 0
Network Memory = 2048m * 0.1 = 205m（而且没超过最大最小值限制）
Direct Memory = 0 + 128m + 205m = 333m
Task Heap = Total Flink Size - Framework Heap - Framework Off-Heap - Task Off-Heap - Managed Memory - Network Memory，即 2048m - 128m - 128m - 0 - 819m - 205m = 768m
Total JVM Heap Size = 768m + 128m = 896m
JVM Metaspace = 256m
（2048m + 256m+ JVM Overhead）*0.1 = JVM Overhead，得到JVM Overhead = 256m
Total Process Size = 2048m + 256m + 256m = 2560m

taskmanager日志上也能看到内存分配值：