flink 内存配置（三）：设置JobManager内存

flink 内存配置（一）：设置Flink进程内存

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

flink 内存配置（三）：设置JobManager内存

flink 内存配置（四）：内存调优和问题处理

JobManager是Flink集群的控制元素。Flink由3部分组成：资源管理器（Resource Manager）、调度器（Dispatcher）和每个运行的Flink作业对应的JobMaster。注意下面的讲解适用于JobManager 1.11之后的版本。

1. JobManager内存模型

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

组件	配置项	描述
JVM Heap	jobmanager.memory.heap.size	JobManager 的 JVM 堆内存大小
Off-heap Memory	jobmanager.memory.off-heap.size	JobManage的堆外内存大小。此选项涵盖所有堆外内存使用，包括直接内存分配和本地内存分配。
JVM metaspace	jobmanager.memory.jvm-metaspace.size	Flink JVM进程的元空间大小
JVM Overhead	jobmanager.memory.jvm-overhead.min jobmanager.memory.jvm-overhead.max jobmanager.memory.jvm-overhead.fraction	为其他JVM开销预留的本机内存：例如线程堆栈、代码缓存、垃圾收集空间等，它是总进程内存的一个受限制的分块组件

2. 配置 JVM Heap

如前面章节 flink 内存配置（一）：设置Flink进程内存里所讲的，为JobManger设置内存的方法，除了直接设置 jobmanager.memory.process.size或 jobmanager.memory.flink.size以外，另外的方法就是设置具体组件的配置，比如这里的JVM配置。

JVM Heap主要用来运行 Flink 框架，执行作业提交时的用户代码以及 Checkpoint 的回调代码等。所需的JVM Heap大小主要由运行作业的数量、它们的结构和用户代码的需求来决定。

注意：如果显示配置了 JVM Heap大小，就不要再去配置 jobmanager.memory.process.size 和 jobmanager.memory.flink.size了，容易引起配置冲突。

Flink脚本和CLI在启动JobManager进程时，可以通过JVM参数-Xms和-Xmx来设置JVM堆大小。

3. 配置Off-heap Memory

堆外内存包含了所有类型的JVM直接内存和本地内存。因此，你还可以通过设置jobmanager.memory.enable-jvm-direct-memory-limit 选项来启用JVM直接内存限制。如果配置了此选项，Flink将通过相应的JVM参数-XX:MaxDirectMemorySize将限制设置为堆外内存大小。

堆外内存可以通过 jobmanager.memory.off-heap.size来设置，这个选项可以被调优，假如 JobManager 抛出 ‘OutOfMemoryError: Direct buffer memory’。

堆外内存的主要消耗用于 Flink框架依赖关系（例如Akka网络通信），执行作业提交时的用户代码以及 Checkpoint 的回调代码等。

注意：如果用户明确配置了Flink总内存（Total Flink Size）和JVM堆（JVM Heap），但没有配置堆外内存，那么堆外内存的大小就是Flink总内存减去JVM堆的大小。堆外内存选项的默认值将被忽略。

4. 本地执行

如果在本地（例如从IDE）运行Flink，而没有创建集群，那么JobManager内存配置选项将被忽略。

5. 实际配置效果展示

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

Total Process Size = Total Flink Size + JVM Metaspace + JVM Overhead
Total Flink Size = Total Heap Size + Total Off-Heap Size
JVM Heap Size = jobmanager.memory.heap.size
Total Off-Heap Size = jobmanager.memory.off-heap.size（默认值128m）
JVM Metaspace = jobmanager.memory.jvm-metaspace.size（默认值256m）
JVM Overhead = Total Process Size * jobmanager.memory.jvm-overhead.fraction（默认值0.1）
其他配置值都可以在官网配置页面查看，注意有些 fraction会有对应的最大最小值限制范围。

情况1：只配置 Total Process Size

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

（这里选择使用application mode 运行在yarn上，温馨提示小心 flink-conf.yaml对命令的影响，导致配置冲突），运行命令：

./bin/flink run-application -t yarn-application -Djobmanager.memory.process.size=2048m ./examples/streaming/TopSpeedWindowing.jar

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

首先只配置了 Total Process Size = 2048m，其他就都是取默认值或推导计算出。
JVM Metaspace = 256m
JVM Overhead = 2048m * 0.1 = 205m
Total Flink Size = 2048m - 256m - 205m = 1587m
JVM Heap没配置，等待被计算出来
Total Off-Heap Size = 128m
JVM Heap = 2048m - 256m - 205 - 128m = 1459m（1.42g），注意上图JVM Heap右边对应的,Metric里不是1.42g，而是1.37g，是因为你使用的 GC 算法会占用其中很小一部分固定内存作为 Non-Heap，可以见下面 Advanced里的 JVM（Heap/Non-Heap）Memory图。

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

情况2：只配置 Total Flink Size

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

（这里选择使用application mode 运行在yarn上），运行命令：

./bin/flink run-application -t yarn-application -Djobmanager.memory.flink.size=2048m ./examples/streaming/TopSpeedWindowing.jar

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

首先只配置了 Total Flink Size = 2048m，其他就都是取默认值或推导计算出。
JVM Metaspace = 256m
Total Off-Heap Size = 128m
JVM Heap = 2048m - 128m = 1920m（1.88g）
Total Process Size = Total Flink Size + JVM Metaspace + JVM Overhead，带入具体数值转化成：Total Process Size = 2048m + 256m + （Total Process Size）*0.1，得到 Total Process Size=2560m
JVM Overhead = 2560m * 0.1 = 256m

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