Flink之TaskManager内存解析

一、CK失败

Flink任务的checkpoint操作失败大致分为两种情况，ck decline和ck expire:

（1）ck decline

发生ck decline情况时，我们可以通过查看JobManager.log或TaskManager.log查明具体原因。其中有一种特殊情况为ck cancel，当前 Flink 中如果较小的ck还没有对齐的情况下，收到了更大的ck，则会把较小的ck给取消掉。

（2）ck expire

如果ck做的非常慢，超过了timeout还没有完成，则整个ck也会失败。这种情况也可以通过查看JobManager.log或TaskManager.log查明具体原因。

由查看JobManager和TaskManager（下文简称TM）当时的日志可知，是因为TM重启，导致做ck超时，发生了ck Expire。

而TM重启的原因，主要有两个原因，一个可能是网络传输波动，另一个是TM资源不足，通过进一步排查，本次TM重启的原因是当时处理数据量增加，导致TM资源不足，发生了TM重启，进而导致了那次ck失败。

二、TaskManager内存分析

TaskManager内存示意图

组成部分	配置参数	描述
框架堆内存（Framework Heap Memory）	taskmanager.memory.framework.heap.size	用于 Flink 框架的 JVM 堆内存
任务堆内存（Task Heap Memory）	taskmanager.memory.task.heap.size	用于 Flink 应用的算子及用户代码的 JVM 堆内存
托管内存（Managed memory）	taskmanager.memory.managed.size	由 Flink 管理的用于排序、哈希表、缓存中间结果及 RocksDB State Backend 的本地内存
框架堆外内存（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	用于任务之间数据传输的直接内存（例如网络传输缓冲）。该内存部分为基于 Flink 总内存的受限的等比内存部分
JVM Metaspace	taskmanager.memory.jvm-metaspace.size	Flink JVM 进程的 Metaspace
JVM 开销	taskmanager.memory.jvm-overhead.min	用于其他 JVM 开销的本地内存，例如栈空间、垃圾回收空间等。该内存部分为基于进程总内存的受限的等比内存部分

Flink并不是将大量对象存在堆上，而是将对象都序列化到一个预分配的内存块上，这个内存块叫做MemorySegment，它代表了一段固定长度的内存（默认大小为32KB），也是Flink中最小的内存分配单元，并且提供了非常高效的读写方法。如果因为内存空间不足，无法申请到更多的内存区域来存储对象时，Flink会将MemorySegment中的数据溢写到本地文件系统（SSD/HDD）中。当再次需要操作数据时，会直接从磁盘中读取数据。

三、调整说明及建议

从以上内容的分析和介绍，在某些情况下，我们可以调整或优化TM的内存，来规避TM重启的问题，最终尽可能避免ck失败的情况。

对于没有硬性资源限制的环境，我们可以使用taskmanager.memory.flink.size参数来配置 Flink总内存的大小，然后Flink自己也会自动根据参数，计算得到各个子区域的配额。如果作业运行正常，则无需单独调整。

如果要更精细化的调整，可以调大JVM Heap中的Task Heap，Task Heap Memory是专门用于执行Flink任务的堆内存空间，是用户代码，自定义数据结构真正占用的内存，通过参数taskmanager.memory.task.heap.size指定。

再其次可以调大Direct Memory中的Task Off-heap Memory，Task Off-heap Memory是Flink执行task所使用的堆外内存。如果在Flink应用的代码中调用了Native的方法，需要用到off-head内存，这些内存会分配到Off-heap堆外内存中，通过参数taskmanage.memory.task.off-heap.size 指定，默认为0。

再其次可以调大Direct Memory中的Network Memory，Flink的Task之间的shuffle，广播等操作以及与外部组件的数据传输需要用到Network Memory，该值通过3个参数确定：

--taskmanager.memory.network.min，Network Memory最小值

--taskmanager.memory.network.max，Network Memory最大值

--taskmanager.memory.network.fraction，Network Memory占Total Flink Memory的比例，默认0.1。如果通过该比例值计算出的结果超出前两个MIN-MAX参数的范围，则以MIN-MAX为准。如果MIN-MAX参数使用同样的值，则表示NetWork是固定的内存大小。

四、可参考的TaskManager内存计算公式

1、每个任务TaskManager分到的总共内存（tm_total_memory）=taskmanager.memory.flink.size - taskmanager.memory.jvm-metaspace.size（JVM元空间，JVM Metaspace）-JVM Overhead Memory （JVM 运行时开销）

其中JVM Overhead Memory用来存放线程栈、编译的代码缓存、JNI 调用的库所分配的内存等等。

--taskmanager.memory.jvm-overhead.fraction，默认 0.1

--taskmanager.memory.jvm-overhead.min，默认 192mb

--taskmanager.memory.jvm-overhead.max，默认 1gb

总进程内存*fraction，如果小于配置的 min（或大于配置的 max）大小，则使用 min/max大小。

2、每个任务TaskManager真正使用的堆内内存(tm_heap_memory)= tm_total_memory- taskmanager.memory.framework.heap.size（堆内框架内存，默认128M - taskmanager.memory.framework.off-heap.size（堆外框架内存,默认128M）- Network Memory（网络内存）- Managed memory（托管内存）

其中Managed Memory托管内存，是有Flink直接管理的堆外内存，用于排序，哈希表，中间结果缓存，以及RocksDB的状态后端。通过参数taskmanage.memory.managed.size指定，默认情况下不配置，通过参数taskmanager.memory.managed.fraction因子(默认0.4) * Total Flink Memory来指定大小。

最后具体情况需要根据业务的复杂度、数据量和集群情况合理分配slot ytm tjm p，其实并行度的设置可以根据算子里面的不同情况各自设置并行度，但是最大的并行度是由 [(slot * jobmanager的数据 ) * nodemanager数量 ]决定的，jobmanager的数量=(可申请的最大内存 - yjm ) / ytm 。其实有的时候slot越大并不会性能越高，集群的资源需要留一部分给hbase hive等数据仓库来做缓存使用，在代码层无法优化后，还是需要根据实际情况测试调整集群资源和运行资源。