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等数据仓库来做缓存使用，在代码层无法优化后，还是需要根据实际情况测试调整集群资源和运行资源。