淄博营销网站建设服务,上海虹口建设局官方网站,做网站需要什么专业方向的员工,聊天软件出售一、CK失败
Flink任务的checkpoint操作失败大致分为两种情况#xff0c;ck decline和ck expire:
#xff08;1#xff09;ck decline
发生ck decline情况时#xff0c;我们可以通过查看JobManager.log或TaskManager.log查明具体原因。其中有一种特殊情况为ck cancelck decline和ck expire:
1ck decline
发生ck decline情况时我们可以通过查看JobManager.log或TaskManager.log查明具体原因。其中有一种特殊情况为ck cancel当前 Flink 中如果较小的ck还没有对齐的情况下收到了更大的ck则会把较小的ck给取消掉。
2ck 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 Memorytaskmanager.memory.framework.heap.size用于 Flink 框架的 JVM 堆内存任务堆内存Task Heap Memorytaskmanager.memory.task.heap.size用于 Flink 应用的算子及用户代码的 JVM 堆内存托管内存Managed memorytaskmanager.memory.managed.size由 Flink 管理的用于排序、哈希表、缓存中间结果及 RocksDB State Backend 的本地内存框架堆外内存Framework Off-heap Memorytaskmanager.memory.framework.off-heap.size用于 Flink 框架的堆外内存直接内存或本地内存任务堆外内存Task Off-heap Memorytaskmanager.memory.task.off-heap.size用于 Flink 应用的算子及用户代码的堆外内存直接内存或本地内存网络内存Network Memorytaskmanager.memory.network.min用于任务之间数据传输的直接内存例如网络传输缓冲。该内存部分为基于 Flink 总内存的受限的等比内存部分JVM Metaspacetaskmanager.memory.jvm-metaspace.sizeFlink JVM 进程的 MetaspaceJVM 开销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 HeapTask Heap Memory是专门用于执行Flink任务的堆内存空间是用户代码自定义数据结构真正占用的内存通过参数taskmanager.memory.task.heap.size指定。
再其次可以调大Direct Memory中的Task Off-heap MemoryTask Off-heap Memory是Flink执行task所使用的堆外内存。如果在Flink应用的代码中调用了Native的方法需要用到off-head内存这些内存会分配到Off-heap堆外内存中通过参数taskmanage.memory.task.off-heap.size 指定默认为0。
再其次可以调大Direct Memory中的Network MemoryFlink的Task之间的shuffle广播等操作以及与外部组件的数据传输需要用到Network Memory该值通过3个参数确定
--taskmanager.memory.network.minNetwork Memory最小值
--taskmanager.memory.network.maxNetwork Memory最大值
--taskmanager.memory.network.fractionNetwork Memory占Total Flink Memory的比例默认0.1。如果通过该比例值计算出的结果超出前两个MIN-MAX参数的范围则以MIN-MAX为准。如果MIN-MAX参数使用同样的值则表示NetWork是固定的内存大小。
四、可参考的TaskManager内存计算公式
1、每个任务TaskManager分到的总共内存tm_total_memorytaskmanager.memory.flink.size - taskmanager.memory.jvm-metaspace.sizeJVM元空间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等数据仓库来做缓存使用在代码层无法优化后还是需要根据实际情况测试调整集群资源和运行资源。
