作者:董伟柯,腾讯 CSIG 高级工程师
概要
Flink 的新版内存管理机制,要追溯到 2020 年初发布的 Flink 1.10 版本。当时 Flink 社区为了实现三大目标:
- 流和批模式下内存管理的统一,即同一套内存配置既可用于流作业也可用于批作业
- 管控好 RocksDB 等外部组件的内存,避免在容器环境下用量不受控导致被 KILL
- 消除不同部署模式下配置参数的歧义,消除 cut-off 等参数语义模糊的问题
提出了两个设计提案 FLIP-49: Unified Memory Configuration for TaskExecutors [1] 和 FLIP-116: Unified Memory Configuration for Job Managers [2],对之前 Flink 内存模型的各项缺陷进行了针对性的重构,为后续的流批一体演进奠定了基础。
由于这个版本距今已有两年多的历史,网上对其内存模型的解读文章也不胜枚举,他们有的对提案进行了中文化的翻译,有的则是对每个参数进行逐一讲解,帮助大家了解 Flink 的内存配置方法 [3] [4]。
本文则是上述简介文章的进一步延展:在新版内存管理模型的基础上,介绍每个区域的技术原理、相关技术资料,以及线上的调优经验,帮助大家在实际应用场景下,更好地规划 Flink 的内存空间,“知其然,也知其所以然”,提前识别和消除隐患。
TaskManager 内存分区总览
我们从 Flink 官网文档的 内存分区图 [5] 开始介绍 ,并加以批注:图的左边标注了每个区域的配置参数名,右边则是一个调优后的、使用 HashMapStateBackend 的作业内存各区域的容量限制:它和默认配置的区别在于 Managed Memory 部分被主动调整为 0,后面我们会讲解何时需要调整各区域的大小,以最大化利用内存空间。
TaskManager 各内存区域详解
接下来,我们详细来看一下各个内存区域的含义、技术原理,以及 Flink 对它的默认值在什么场景下需要调整。
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 而导致的作业频繁重启。
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),我们接下来会讲到它。
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)的比例。我们接下来也会讲到它。
JVM 堆外内存(JVM Off-Heap Memory)
广义上的 堆外内存 指的是 JVM 堆之外的内存空间,而我们这里特指 JVM 进程总内存除了元空间(Metaspace)和运行时开销(Overhead)以外的内存区域。因为上述两个区域是 JVM 自行管理,Flink 无法介入,我们后面单独划分和讲解。
托管内存(Managed Memory)
文章开头的总览图中,把托管内存区域设为 0,此时任务堆空间约 3G;而使用 Flink 默认配置时,任务堆只有 1.5G。这是因为默认情况下,托管内存占了 40% 的 Flink 总内存,导致堆内存可用的量变的相当少。因此我们非常有必要了解什么是托管内存。
从官方文档和 Flink 源码上来看,托管内存主要有三大使用场景:
- 批处理算法,例如排序、HashJoin 等。他们会从 Flink 的 MemoryManager 请求内存片段(MemorySegment),而 MemoryManager 则会调用
UNSAFE.allocateMemory
分配堆外内存。 - RocksDB StateBackend,Flink 只会预留一部分空间并扣除预算,但是不介入实际内存分配。因此该类型的内存资源被称为
OpaqueMemoryResource
. 实际的内存分配还是由 JNI 调用的 RocksDB 自己通过 malloc 函数申请。 - PyFlink。与 JNI 类似,在与 Python 进程交互的过程中,也会用到一部分托管内存。
显然,对于普通的流式 SQL 作业,如果启用了 RocksDB 状态后端时,才会大量使用托管内存。因此如果您的业务场景并未用到 RocksDB,那么可以调小托管内存的相对比例(taskmanager.memory.managed.fraction
)或绝对大小(taskmanager.memory.managed.size
),以增大任务堆的空间。
对于 RocksDB 作业,之所以分配了 40% Flink 总内存,是因为 RocksDB 的内存用量实在是一个很头疼的问题。早在 2017 年,就有 FLINK-7289: Memory allocation of RocksDB can be problematic in container environments [6] 这个问题单,随后社区对此做了大量的工作(通过 LRUCache 参数、增强 WriteBufferManager 的 Slot 内空间复用等),来尽可能地限制 RocksDB 的总内存用量。在我之前的 Flink on RocksDB 参数调优指南 [7] 文章中,也有提到 RocksDB 内存调优的各项参数,其中 MemTable、Block Cache 都是托管内存空间的用量大户。
为了避免手动调优的繁杂,Flink 新版内存管理默认将 state.backend.rocksdb.memory.managed
参数设为 true
,这样就由 Flink 来计算 RocksDB 各部分需要用多少内存 [8],这也是 ”托管“ 的含义所在。如果仍然希望精细化手动调整 RocksDB 参数,则需要将上述参数设为 false
.
直接内存(Direct Memory)
直接内存是 JVM 堆外的一类内存,它提供了相对安全可控但又不受 GC 影响的空间,JVM 参数是 -XX:MaxDirectMemorySize
. 它主要用于
- 框架自身(
taskmanager.memory.framework.off-heap.size
参数,默认 128M,例如 Sort-Merge Shuffle 算法所需的内存) - 用户任务(
taskmanager.memory.task.off-heap.size
参数,默认设为 0) - 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 不足导致作业报错。
JVM 元空间(JVM Metaspace)
JVM Metaspace 主要保存了加载的类和方法的元数据,Flink 配置的参数是 taskmanager.memory.jvm-metaspace.size
,默认大小为 256M,JVM 参数是 -XX:MaxMetaspaceSize
.
如果用户编写的 Flink 程序中,有大量的动态类加载的需求,例如我们之前遇到过一个用户作业,动态编译并加载了 44 万个类,此时就容易出现元空间用量远超预期,发生 OOM 报错。此时就需要适当调大元空间的大小,或者优化用户程序,及时卸载无用的 Classloader。
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 区空间,以留出更多的安全余量。
参考阅读
[1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-49: Unified Memory Configuration for TaskExecutors
[2] https://cwiki.apache.org/confluence/display/FLINK/FLIP-116: Unified Memory Configuration for Job Managers
[3] https://www.jianshu.com/p/96364463c831
[4] https://zhuanlan.zhihu.com/p/141120042
[5] https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/memory/mem_setup_tm/#内存模型详解
[6] https://issues.apache.org/jira/browse/FLINK-7289
[7] https://cloud.tencent.com/developer/article/1592441
[8] https://www.jianshu.com/p/47a40259a450
[9] https://issues.apache.org/jira/browse/FLINK-15532
扫码加入 流计算 Oceanus 产品交流群