flink内存管理，设置思路，oom问题，一文全_flink managed memory

这里以flink1.12 的 flink webUI 来展示内存管理，后续版本的内存可能会有变更不一致的地方，详细的解释主要放在taskManager中

1 内存分配

下图的左边标注了每个区域的配置参数名，右边则是一个调优后的、使用 HashMapStateBackend 的作业内存各区域的容量限制：它和默认配置的区别在于 Managed Memory 部分被主动调整为 0，后面我们会讲解何时需要调整各区域的大小，以最大化利用内存空间。

1.1 JVM 进程总内存（Total Process Memory）

• 该区域表示在容器环境下，TaskManager 所在 JVM 的最大可用的内存配额，包含了本文后续介绍的所有内存区域，超用时可能被强制结束进程。我们可以通过 taskmanager.memory.process.size 参数控制它的大小。
• 例如我们设置 JVM 进程总内存为 4G，TaskManager 运行在 Kubernetes 平台，则 Pod 配置的 spec -> resources -> limits -> memory 项会被设置为 4Gi，源码见 org.apache.flink.kubernetes.kubeclient.decorators.InitTaskManagerDecorator#decorateMainContainer，运行时的 YAML 配置如下图：
  
• 而对于 YARN，如果 yarn.nodemanager.pmem-check-enabled 设为 true, 则也会在运行时定期检查容器内的进程是否超用内存。
• 如果进程总内存用量超出配额，容器平台通常会直接发送最严格的 SIGKILL 信号（相当于 kill -9）来中止 TaskManager，此时不会有任何延期退出的机会，可能会造成作业崩溃重启、外部系统资源无法释放等严重后果。
• 因此，在 有硬性资源配额检查 的容器环境下，请务必妥善设置该参数，对作业充分压测后，尽可能预留一部分安全余量，避免 TaskManager 频繁被 KILL 而导致的作业频繁重启。

1.2 Flink 总内存（Total Flink Memory）

• 该内存区域指的是 Flink 可以控制的内存区域，即上述提到的 JVM 进程总内存 减去 Flink 无法控制的 Metaspace（元空间）和 Overhead（运行时开销）区域。Flink 随后又把这部分内存区域划分为堆内、堆外（Direct）、堆外（Managed）等不同子区域，后面我们会逐一讲解他们的配置指南。
• 对于没有硬性资源限制的环境，我们建议使用 taskmanager.memory.flink.size 参数来配置 Flink 总内存的大小，然后 Flink 自己也会会自动根据参数，计算得到各个子区域的配额。如果作业运行正常，则无需单独调整。
• 例如 4G 的 进程总内存 配置下，JVM 运行时开销（Overhead）占 进程总内存 的 10% 但最多 1G（下图是 409.6M），元空间（Metaspace）占 256M；堆外直接（Direct）内存网络缓存占 Flink 总内存 的 10% 但最多 1G（下图是 343M），框架堆和框架堆外各占 128M，堆外管控（Managed）内存占 Flink 总内存 的 40%（下图是 1372M 即 1.34G），其他空间留给任务堆，即用户程序代码可以使用的内存空间（1459M 即 1.42G），我们接下来会讲到它。
  

1.3 JVM 堆外内存（JVM Off-Heap Memory）

• 广义上的 堆外内存 指的是 JVM 堆之外的内存空间，而我们这里特指 JVM 进程总内存除了元空间（Metaspace）和运行时开销（Overhead）以外的内存区域。因为上述两个区域是 JVM 自行管理，Flink 无法介入，我们后面单独划分和讲解。

1.4 JVM 堆内存（JVM Heap Memory）

• 堆内存大家想必都不陌生，它是由 JVM 提供给用户程序运行的内存区域，JVM 会按需运行 GC（垃圾回收器），协助清理失效对象。
• 当任务启动时，ProcessMemoryUtils#generateJvmParametersStr 方法会通过 -Xmx -Xms 参数设置堆内存的最大容量。
• Flink 将堆内存从逻辑上划分为 “框架堆”、“任务堆” 两个子区域，分别通过 taskmanager.memory.framework.heap.size 和 taskmanager.memory.task.heap.size 来指定其大小：框架堆默认是 128m，任务堆如果未显式设置其大小，则会通过扣减其他区域配额来计算得到。例如对于 4G 的进程总内存，扣除了其他区域后，任务堆可用的只有不到 1.5G。
• 但需要注意的是，Flink 自身并不能精确控制框架自身及任务会用多少堆内存，因此上述配置项只提供理论上的计算依据。如果实际用量超出配额，且 JVM 难以回收对象释放空间，则会抛出 OutOfMemoryError，此时 Flink TaskManager 会退出，导致作业崩溃重启。因此对于堆内存的监控是必须要配置的，当堆内存用量超过一定比率，或者 Full GC 时长和次数明显增长时，需要尽快介入并考虑扩容。
• 高级内容：对于使用 HashMapStateBackend（旧版本称之为 FileSystem StateBackend）的流作业用户，如果在进程总内存固定的前提下，希望尽可能提升任务堆的空间，则可以减少 托管内存（Managed Memory）的比例。我们接下来也会讲到它。

1.5 托管内存（Managed Memory）

• 文章开头的总览图中，把托管内存区域设为 0，此时任务堆空间约 3G；而使用 Flink 默认配置时，任务堆只有 1.5G。这是因为默认情况下，托管内存占了 40% 的 Flink 总内存，导致堆内存可用的量变的相当少。因此我们非常有必要了解什么是托管内存。

从官方文档和 Flink 源码上来看，托管内存主要有三大使用场景：

1. 批处理算法，例如排序、HashJoin 等。他们会从 Flink 的 MemoryManager 请求内存片段（MemorySegment），而 MemoryManager 则会调用 UNSAFE.allocateMemory 分配堆外内存。
2. RocksDB StateBackend，Flink 只会预留一部分空间并扣除预算，但是不介入实际内存分配。因此该类型的内存资源被称为 OpaqueMemoryResource. 实际的内存分配还是由 JNI 调用的 RocksDB 自己通过 malloc 函数申请。
3. PyFlink。与 JNI 类似，在与 Python 进程交互的过程中，也会用到一部分托管内存。
4. 显然，对于普通的流式 SQL 作业，如果启用了 RocksDB 状态后端时，才会大量使用托管内存。因此如果您的业务场景并未用到 RocksDB，那么可以调小托管内存的相对比例（taskmanager.memory.managed.fraction）或绝对大小（taskmanager.memory.managed.size），以增大任务堆的空间。
5. 对于 RocksDB 作业，之所以分配了 40% Flink 总内存，是因为 RocksDB 的内存用量实在是一个很头疼的问题。早在 2017 年，就有 FLINK-7289: Memory allocation of RocksDB can be problematic in container environments [6] 这个问题单，随后社区对此做了大量的工作（通过 LRUCache 参数、增强 WriteBufferManager 的 Slot 内空间复用等），来尽可能地限制 RocksDB 的总内存用量。有些文章中，也有提到 RocksDB 内存调优的各项参数，其中 MemTable、Block Cache 都是托管内存空间的用量大户。
6. 为了避免手动调优的繁杂，Flink 新版内存管理默认将 state.backend.rocksdb.memory.managed 参数设为 true，这样就由 Flink 来计算 RocksDB 各部分需要用多少内存 [8]，这也是 ”托管“ 的含义所在。如果仍然希望精细化手动调整 RocksDB 参数，则需要将上述参数设为 false.

1.6直接内存（Direct Memory）

直接内存是 JVM 堆外的一类内存，它提供了相对安全可控但又不受 GC 影响的空间，JVM 参数是 -XX:MaxDirectMemorySize. 它主要用于

1. 框架自身（taskmanager.memory.framework.off-heap.size 参数，默认 128M，例如 Sort-Merge Shuffle 算法所需的内存）
2. 用户任务（taskmanager.memory.task.off-heap.size 参数，默认设为 0）
3. Netty 对 Network Buffer 的网络传输（taskmanager.memory.network.fraction 等参数，默认 0.1 即 10% 的 Flink 总内存）。

• 在生产环境中，如果作业并行度非常大（例如大于 500 甚至 1000），则需要调大 taskmanager.network.memory.floating-buffers-per-gate（例如从 8 调整到 1000）和 taskmanager.network.memory.buffers-per-channel（例如从 2 调整到 500），避免 Network Buffer 不足导致作业报错。

1.7 JVM 元空间（JVM Metaspace）

• JVM Metaspace 主要保存了加载的类和方法的元数据，Flink 配置的参数是 taskmanager.memory.jvm-metaspace.size，默认大小为 256M，JVM 参数是 -XX:MaxMetaspaceSize.
• 如果用户编写的 Flink 程序中，有大量的动态类加载的需求，例如我们之前遇到过一个用户作业，动态编译并加载了 44 万个类，此时就容易出现元空间用量远超预期，发生 OOM 报错。此时就需要适当调大元空间的大小，或者优化用户程序，及时卸载无用的 Classloader。

1.8 JVM 运行时开销（JVM Overhead）

• 除了上述描述的内存区域外，JVM 自己还有一小块 “自留地”，用来存放线程栈、编译的代码缓存、JNI 调用的库所分配的内存等等，Flink 配置参数是 taskmanager.memory.jvm-overhead.fraction，默认是 JVM 总内存的 10%。
• 对于旧版本（1.9 及之前）的 Flink，RocksDB 通过 malloc 分配的内存也属于 Overhead 部分，而新版 Flink 把这部分归类到托管 内存（Managed），但由于 FLINK-15532 Enable strict capacity limit for memory usage for RocksDB [9] 问题仍未解决，RocksDB 仍然会少量超用一部分内存。
• 因此在生产环境下，如果 RocksDB 频繁造成内存超用，除了调大 Managed 托管内存外，也可以考虑调大 Overhead 区空间，以留出更多的安全余量。

来自flink 1.12的per-job模式下jobmanager的 内存分配

jvm metaspace存放jvm加载的类的元数据
还有一些想 Network 这种网络开销，一半不涉及修改的内存，没有标出来。

内存数据结构：
flink最小内存分配单元 是内存段MemorySegment 默认32k。

2 内存设置思路

2.1 配置举例

在yarn提交名利中 可以对yarn进行配置来修改内存大小
如使用的是fs memory state，并且没有什么哈希排序操作，就可以修改managed Memory的大小，让更多的资源运用于task的堆上内存中
-yD taskmanager.memory.size=8192 \
-yD taskmanager.memory.fraction=0.3
-yD taskmanager.memory.managed.size=0.3
具体参数可以参考flink官网 ：
https://nightlies.apache.org/flink/flink-docs-release-1.3/setup/config.html#managed-memory

2.2 配置思路

2.2.1 并行度，slot，taskmanager数量三者的关系

在这里我们先说一下基本概念：
 用户通过算子 api 所开发的代码，会被 flink 任务提交客户端解析成 jobGraph
 然后，jobGraph 提交到集群 JobManager，转化成 ExecutionGraph（并行化后的执行图）
 然后，ExecutionGraph 中的各个 task 会以多并行实例（subTask）部署到 taskmanager 上执行；
 subTask 运行的位置是 taskmanager 所提供的槽位（task slot），槽位简单理解就是线程；
一个算子的逻辑，可以封装在一个独立的 task 中（可以有多个运行时实例：subTask）；
也可把多个算子的逻辑 chain 在一起后封装在一个独立的 task 中（可以有多个运行时实例：subTask）；

简单点说 并行度=slot*taskmanager ；
例如下图，2个tm 2个 slot 并行度就是4

在 yarn 中提交一个 flink 任务, container 数量计算方式如下
container.num == taskmanager.num == ( parallelism.default / taskmanager.numberOfTaskSlots )

parallelism 是指 taskmanager 实际使用的并发能力。假设我们把 parallelism.default 设置
为 1，那么 9 个 TaskSlot 只能用 1 个，有 8 个空闲。 并发数<=slot*tm数 slot不够就会自动起tm，来补充

设置parallelism有多中方式，优先级为api>env>p>file

2.2.2 资源大小选择

2.2.2.1 举例：

假设设置单个taskmanager为14g，taskmanager.memory.managed.fraction为0.5，将会得到以下内容：

JVM Heap Size：5.19 GB Flink Managed Memory：6.45 GB

JVM (Heap/Non-Heap) ： Heap：5.19 GB Non-Heap：1.33 GB Total：6.52 GB

Outside JVM：Capacity：1.01GB

NetWork: count: xxxxx

可以计算得到6.45+6.52+1.01 = 13.98 等于14

2.2.2.1 计算过程

taskmanager.memory.process.size 设置的是容器的内存大小。
计算过程在org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils中processSpecFromConfig方法，TaskExecutorProcessSpec类展示了1.10版本整个内存的组成。

计算方法分成3种：

1. 指定了taskmanager.memory.task.heap.size和taskmanager.memory.managed.size 见方法：deriveProcessSpecWithExplicitTaskAndManagedMemory
2. 指定了taskmanager.memory.flink.size 见方法：deriveProcessSpecWithTotalFlinkMemory
3. 指定了taskmanager.memory.process.size（容器环境一般指定这个，决定全局容量）

totalProcessMemorySize = 设置的值 14g
jvmMetaspaceSize = taskmanager.memory.jvm-metaspace.size 默认96m
这个对应参数-XX:MaxMetaspaceSize=100663296。

jvmOverheadSize：

• taskmanager.memory.jvm-overhead.min 192m
• taskmanager.memory.jvm-overhead.max 1g
• taskmanager.memory.jvm-overhead.fraction 0.1

公式 14g * 0.1 = 1.4g 必须在[192m, 1g]之间，所以jvmOverheadSize的大小是1g

• totalFlinkMemorySize = 14g - 1g - 96m = 13216m

• frameworkHeapMemorySize：taskmanager.memory.framework.heap.size 默认128m

• frameworkOffHeapMemorySize：taskmanager.memory.framework.off-heap.size 默认128m

• taskOffHeapMemorySize：taskmanager.memory.task.off-heap.size 默认0

确定好上面这些参数后，就是最重要的三个指标的计算了：taskHeapMemorySize，networkMemorySize，managedMemorySize

计算分成确定了：taskmanager.memory.task.heap.size还是没确定。

1）确定taskmanager.memory.task.heap.size

• taskHeapMemorySize = 设置值

• managedMemorySize = 设置了使用设置值，否则使用 0.4 * totalFlinkMemorySize

• 如果 taskHeapMemorySize + taskOffHeapMemorySize + frameworkHeapMemorySize + frameworkOffHeapMemorySize + managedMemorySize > totalFlinkMemorySize异常

• networkMemorySize 等于剩余的大小，之后还会check这块内存是否充足，可以自己查看对应代码

2）未设置heap大小

• 先确定 managedMemorySize = 设置了使用设置值，否则使用 0.4 * totalFlinkMemorySize，这里就是 0.5 * 13216m = 6608 = 6.45g (这里就是dashboard的显示内容)

• 再确定network buffer大小，这个也是有两种情况，不细说。 [64mb, 1g] 0.1 * totalFlinkMemorySize = 1321.6, 所以是1g

• 最后剩余的就是taskHeapMemorySize,不能为负数，这里等于 13216 - 6608 - 1024 - 128 - 128 = 5328 = 5.2g (这里约等于dashboard的显示heap大小)

3）最后jvm的参数的计算过程：

• jvmHeapSize = frameworkHeapSize + taskHeapSize = 5328 + 128 = 5456

• jvmDirectSize = frameworkOffHeapMemorySize + taskOffHeapSize + networkMemSize = 128 + 1024 = 1152

• jvmMetaspaceSize = 96m

3.OOM Killed 常见原因

本部分内容转载自：
本文作者：林小铂 (Paul Lin)
本文链接： 2021/01/02/详解-Flink-容器化环境下的-OOM-Killed/
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 CN 许可协议。转载请注明出处！

实践中导致 OOM Killed 的常见原因基本源于 Native 内存的泄漏或者过度使用。因为虚拟内存的 OOM Killed 通过资源管理器的配置很容易避免且通常不会有太大问题，所以下文只讨论物理内存的 OOM Killed。
从表中可以看到，使用 Heap、Metaspace 和 Direct 内存都是比较安全的，但非 Direct 的 Native 内存情况则比较复杂，可能是 JVM 本身的一些内部使用（比如下文会提到的 MemberNameTable），也可能是用户代码引入的 JNI 依赖，还有可能是用户代码自身通过 sun.misc.Unsafe 申请的 Native 内存。理论上讲，用户代码或第三方 lib 申请的 Native 内存需要用户来规划内存用量，而 Internal 的其余部分可以并入 JVM 本身的内存消耗。而实际上 Flink 的内存模型也遵循了类似的原则。

3.1 RocksDB Native 内存的不确定性

众所周知，RocksDB 通过 JNI 直接申请 Native 内存，并不受 Flink 的管控，所以实际上 Flink 通过设置 RocksDB 的内存参数间接影响其内存使用。然而，目前 Flink 是通过估算得出这些参数，并不是非常精确的值，其中有以下的几个原因。

首先是部分内存难以准确计算的问题。RocksDB 的内存占用有 4 个部分:

• Block Cache: OS PageCache 之上的一层缓存，缓存未压缩的数据 Block。

• Indexes and filter blocks: 索引及布尔过滤器，用于优化读性能。

• Memtable: 类似写缓存。

• Blocks pinned by Iterator: 触发 RocksDB 遍历操作（比如遍历 RocksDBMapState 的所有 key）时，Iterator 在其生命周期内会阻止其引用到的 Block 和 Memtable 被释放，导致额外的内存占用。

前三个区域的内存都是可配置的，但 Iterator 锁定的资源则要取决于应用业务使用模式，且没有提供一个硬限制，因此 Flink 在计算 RocksDB StateBackend 内存时没有将这部分纳入考虑。

其次是 RocksDB Block Cache 的一个 bug，它会导致 Cache 大小无法严格控制，有可能短时间内超出设置的内存容量，相当于软限制。

对于这个问题，通常我们只要调大 JVM Overhead 的阈值，让 Flink 预留更多内存即可，因为 RocksDB 的内存超额使用只是暂时的。

3.2 Glibc Thread Arena 问题

另外一个常见的问题就是 glibc 著名的 64 MB 问题，它可能会导致 JVM 进程的内存使用大幅增长，最终被 YARN kill 掉。

具体来说，JVM 通过 glibc 申请内存，而为了提高内存分配效率和减少内存碎片，glibc 会维护称为 Arena 的内存池，包括一个共享的 Main Arena 和线程级别的 Thread Arena。当一个线程需要申请内存但 Main Arena 已经被其他线程加锁时，glibc 会分配一个大约 64 MB (64 位机器)的 Thread Arena 供线程使用。这些 Thread Arena 对于 JVM 是透明的，但会被算进进程的总体虚拟内存（VIRT）和物理内存（RSS）里。

默认情况下，Arena 的最大数目是 cpu 核数 * 8，对于一台普通的 32 核服务器来说最多占用 16 GB，不可谓不可观。为了控制总体消耗内存的总量，glibc 提供了环境变量 MALLOC_ARENA_MAX 来限制 Arena 的总量，比如 Hadoop 就默认将这个值设置为 4。然而，这个参数只是一个软限制，所有 Arena 都被加锁时，glibc 仍会新建 Thread Arena 来分配内存，造成意外的内存使用。

通常来说，这个问题会出现在需要频繁创建线程的应用里，比如 HDFS Client 会为每个正在写入的文件新建一个 DataStreamer 线程，所以比较容易遇到 Thread Arena 的问题。如果怀疑你的 Flink 应用遇到这个问题，比较简单的验证方法就是看进程的 pmap 是否存在很多大小为 64MB 倍数的连续 anon 段，比如下图中蓝色几个的 65536 KB 的段就很有可能是 Arena。

这个问题的修复办法比较简单，将 MALLOC_ARENA_MAX 设置为 1 即可，也就是禁用 Thread Arena 只使用 Main Arena。当然，这样的代价就是线程分配内存效率会降低。不过值得一提的是，使用 Flink 的进程环境变量参数（比如 containerized.taskmanager.env.MALLOC_ARENA_MAX=1）来覆盖默认的 MALLOC_ARENA_MAX 参数可能是不可行的，原因是在非白名单变量（yarn.nodemanager.env-whitelist）冲突的情况下， NodeManager 会以合并 URL 的方式来合并原有的值和追加的值，最终造成 MALLOC_ARENA_MAX=“4:1” 这样的结果。

最后，还有一个更彻底的可选解决方案，就是将 glibc 替换为 Google 家的 tcmalloc 或 Facebook 家的 jemalloc [12]。除了不会有 Thread Arena 问题，内存分配性能更好，碎片更少。在实际上，Flink 1.12 的官方镜像也将默认的内存分配器从 glibc 改为 jemelloc 。

3.3 JDK8 Native 内存泄漏

Oracle Jdk8u152 之前的版本存在一个 Native 内存泄漏的 bug[13]，会造成 JVM 的 Internal 内存分区一直增长。

具体而言，JVM 会缓存字符串符号（Symbol）到方法（Method）、成员变量（Field）的映射对来加快查找，每对映射称为 MemberName，整个映射关系称为 MemeberNameTable，由 java.lang.invoke.MethodHandles 这个类负责。在 Jdk8u152 之前，MemberNameTable 是使用 Native 内存的，因此一些过时的 MemberName 不会被 GC 自动清理，造成内存泄漏。

要确认这个问题，需要通过 NMT 来查看 JVM 内存情况，比如笔者就遇到过线上一个 TaskManager 的超过 400 MB 的 MemeberNameTable。

在 JDK-8013267[14] 以后，MemeberNameTable 从 Native 内存被移到 Java Heap 当中，修复了这个问题。然而，JVM 的 Native 内存泄漏问题不止一个，比如 C2 编译器的内存泄漏问题[15]，所以对于跟笔者一样没有专门 JVM 团队的用户来说，升级到最新版本的 JDK 是修复问题的最好办法。

3.4 YARN mmap 内存算法

众所周知，YARN 会根据 /proc/${pid} 下的进程信息来计算整个 container 进程树的总体内存，但这里面有一个比较特殊的点是 mmap 的共享内存。mmap 内存会全部被算进进程的 VIRT，这点应该没有疑问，但关于 RSS 的计算则有不同标准。

依据 YARN 和 Linux smaps 的计算规则，内存页（Pages）按两种标准划分:

• Private Pages: 只有当前进程映射（mapped）的 Pages
• Shared Pages: 与其他进程共享的 Pages
• Clean Pages: 自从被映射后没有被修改过的 Pages
• Dirty Pages: 自从被映射后已经被修改过的 Pages

在默认的实现里，YARN 根据 /proc/${pid}/status 来计算总内存，所有的 Shared Pages 都会被算入进程的 RSS，即便这些 Pages 同时被多个进程映射[16]，这会导致和实际操作系统物理内存的偏差，有可能导致 Flink 进程被误杀（当然，前提是用户代码使用 mmap 且没有预留足够空间）。

为此，YARN 提供 yarn.nodemanager.container-monitor.procfs-tree.smaps-based-rss.enabled 配置选项，将其设置为 true 后，YARN 将根据更准确的 /proc/${pid}/smap 来计算内存占用，其中很关键的一个概念是 PSS。简单来说，PSS 的不同点在于计算内存时会将 Shared Pages 均分给所有使用这个 Pages 的进程，比如一个进程持有 1000 个 Private Pages 和 1000 个会分享给另外一个进程的 Shared Pages，那么该进程的总 Page 数就是 1500。

回到 YARN 的内存计算上，进程 RSS 等于其映射的所有 Pages RSS 的总和。在默认情况下，YARN 计算一个 Page RSS 公式为:

Page RSS = Private_Clean + Private_Dirty + Shared_Clean + Shared_Dirty

因为一个 Page 要么是 Private，要么是 Shared，且要么是 Clean 要么是 Dirty，所以其实上述公示右边有至少三项为 0 。而在开启 smaps 选项后，公式变为:

Page RSS = Min(Shared_Dirty, PSS) + Private_Clean + Private_Dirty

简单来说，新公式的结果就是去除了 Shared_Clean 部分被重复计算的影响。

虽然开启基于 smaps 计算的选项会让计算更加准确，但会引入遍历 Pages 计算内存总和的开销，不如 直接取 /proc/${pid}/status 的统计数据快，因此如果遇到 mmap 的问题，还是推荐通过提高 Flink 的 JVM Overhead 分区容量来解决。