Flink 内存模型和生产环境调优

Flink 自1.1版本起通过 FLIP-49 Unified Memory Configuration for TaskExecutors 和 FLIP-116 Unified Memory Configuration for Job Managers对 TaskManager 和 JobManager 的内存模型进行了重大重构，统一了内存配置方式，本文详细介绍 Flink TaskManager 内存模型的各内存块配置参数、生产环境中的优化策略，以及 Jemalloc 内存分配器的应用。

Flink 内存模型结构图

图片

On-Heap [JVM 堆内存] 1、Framework Heap Memory: Flink 框架本身使用的内存，即 TaskManager 本身所占用的堆内存。

配置参数:

taskmanager.memory.framework.heap.size=128MB，默认 128MB

2、Task Heap Memory: Task 执行用户代码时所使用的堆内存。

配置参数:

taskmanager.memory.task.heap.size。

Off-Heap [JVM 堆外内存]

1、DirectMemory

1.1. Framework Off-Heap Memory: Flink框架本身所使用的内存，即TaskManager 本身所占用的堆外内存。

配置参数:

taskmanager.memory.framework.off-heap.size=128MB，默认 128MB

1.2. Task Off-Heap Memory: Task 执行用户代码所使用的堆外内存。

配置参数:

taskmanager.memory.task.off-heap.size=0，默认 0

1.3. Network Memory: 网络数据交换所使用的堆外内存大小,如网络数据交换。

配置参数:

taskmanager.memory.network.fraction=0.1

taskmanager.memory.network.min=64MB，taskmanager.memory.network.max=1GB

2、Managed Memory: Flink 管理的堆外内存,用于排序、哈希表、缓存中间结果及 RocksDB State Backend 的本地内存。

配置参数:

taskmanager.memory.managed.fraction=0.4

taskmanager.memory.managed.size=xx

3、JVM metaspace: 用于存储类的元数据信息

配置参数:

taskmanager.memory.jvm-metaspace.size=128MB

4、JVM over-head 执行开销:JVM 执行时自身所需要的内容,包括线程堆栈、IO、 编译缓存等所使用的内存。

配置参数:

taskmanager.memory.jvm-overhead.min=192MB, taskmanager.memory.jvm-overhead.max=512MB, taskmanager.memory.jvm-overhead.fraction=0.1

总体内存 1、总进程内存: Flink Java 应用程序(包括用户代码)和 JVM 运行整个进程所消耗的总内存。 总进程内存 = Flink 使用内存 + JVM metaspace + JVM Overhead

配置参数:

taskmanager.memory.process.size=xxx 2、Flink 总内存: 仅 Flink Java 应用程序消耗的内存，包括用户代码，但不包括 JVM 为其运行而分配的内存

Flink 使用内存 = 框架堆内外 + task 堆内外 + network + manage

配置参数:

taskmanager.memory.flink.size=xxx

对应调优策略

内存调优是优化 Flink 作业性能的重要步骤，合理的内存配置可以减少 GC 压力、提高吞吐量并避免作业失败；相反如果对作业设置的内存不合理，可能导致内存资源浪费或任务出现问题。

图片

1.Task Heap内存

可以通过 Taskmanager 的 Full GC 时长和次数进行调整。频繁 Full GC 则适当调整这块内存，反之可以考虑调小这块内存，通过 TM Process 内存调整，也可以使用 G1 垃圾回收器（-XX:+UseG1GC）Full GC 频率。

2.Managed Memory内存

如果任务不存在窗口或者自定义state(基于rocksdb)，表象为所有 Taskmanager 的这个 Metric 指标数据都是0%，这时候可以把该内存去掉，对应值设置为0。如果有基于 rocksdb 的状态数据，可以设置 state.backend.rocksdb.memory.managed=true，交由Flink托管，如果对 rocksdb 熟悉，想自己进行精细化调优，则设置为 false，并调优 rocksdb 相关参数。

通过自定义系统参数进行修改：

-Dtaskmanager.memory.managed.size=xxMB

3.Task Off-Heap内存

此块内存默认配置值是0，一般不需要用户手动去调整，如果用户的代码程序里用到了 DirectMemory，或者依赖了第三方框架例如 Pulsar, Netty，对 DirectMemory 有需求，那么用户需要手动指定这块内存的大小，不然容易出现的 java.lang.OutOfMemoryError: Direct buffer memory 问题

通过自定义系统参数进行修改：

-Dtaskmanager.memory.task.off-heap.size=xxMB

4.Network内存

Network 内存跟任务的整体并行度以及执行图的复杂情况有关，如果任务并行度较大导致的这块内存不够，会报一下异常，这时候需要适当的提升网络内存的大小。一般并行度较小或执行图简单的任务，可以把这块内存调下。

java.io.IOException: Insufficient number of network buffers: required xx, but only 0 available.  

-Dtaskmanager.memory.network.min=xxMB

-Dtaskmanager.memory.network.max=xxMB

5.JVM Metaspace内存

如果这块内存大于 128MB，可以尝试把这块内存调到 128MB。因为大部分作业 128MB 足够 Metaspace 使用。

通过自定义系统参数进行修改：

-Dtaskmanager.memory.jvm-metaspace.size=128M

6.JVM Overhead内存

如果使用rocksdb状态后端并且有内存超用现象，可以适当调大这块内存。

通过自定义系统参数进行修改：

-Dtaskmanager.memory.jvm-overhead.max=xxMB

-Dtaskmanager.memory.jvm-overhead.min=xxMB

Jemalloc内存分配器

jemalloc 是一种高效的内存分配器，广泛用于替代标准 glibc 内存分配器，能够减少内存碎片、提高分配效率并降低延迟，通过jemalloc内存分配器也能优化 Flink 的内存使用。在 Flink 的 YARN 部署环境中，通过指定-Dyarn.ship-archives和-Dcontainerized.taskmanager.env.LD_PRELOAD 的方式来使用 jemalloc，这种方式允许在 YARN 容器中分发和加载 jemalloc 库，替代默认的 glibc 内存分配器，以优化 Flink TaskManager 的内存使用。

那么，使用 jemalloc 优化的又是哪块内存呢？

更多大数据干货，欢迎关注我的微信公众号—BigData共享