Flink Checkpoint 超时原因及解决

╰半橙微兮° 2021-09-25 07:00 2274阅读 0赞

### 第一种、计算量大，CPU密集性，导致TM内线程一直在processElement，而没有时间做CP【过滤掉部分数据；增大并行度】 ###

代表性作业为算法指标-用户偏好的计算，需要对用户在商城的曝光、点击、订单、出价、上下滑等所有事件进行比例计算，并且对各个偏好值进行比例计算，事件时间范围为近24小时。等于说每来一条数据，都需要对用户近24小时内所有的行为事件进行分类汇总，求比例，再汇总，再求比例，而QPS是1500，24小时1.5亿的累积数据，逻辑处理的算子根本无法将接收到的数据在合适的时间内计算完毕，这里还有个有趣的现象，为了提高处理性能，我将并行度翻倍，结果checkpoint的时间反而更长了，原因是Source的并行度也增加后，读取源数据的速度更快了～  
从图片中可以看到source、sink的cp时间都很快，只有处理节点的‘End to End Duration’时间特别长，其他的‘Checkpoint Duration (Sync)’、‘Checkpoint Duration (Async)’时间都很短，都为几百毫秒。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1bGVpZGFyZW4_size_16_color_FFFFFF_t_70_pic_center]  
那么怎么办呢？这里我也反思了自己的实现逻辑，实时计算中，flink是流引擎，正确的使用姿势应该是对每一条数据进行实时处理，而不应该对较长历史时间范围内的历史数据进行批处理，如果每条数据来还需要对历史数据重新计算计算，那么就不符合flink的定位。所以和算法同学商议后，将实现逻辑进行修改，进行批流分开计算，比如离线数据每半个小时进行一次计算，而实时计算只需要计算最近半小时内的数据即可。总之两个方法，一、减少源数据量，过滤黑名单或者非法ID；window聚合； 二、简化处理逻辑，特别是减少遍历。

### 第二种、数据倾斜【[https://programskills.blog.csdn.net/article/details/115641985][https_programskills.blog.csdn.net_article_details_115641985]解决方法】 ###

代表性作业对手机的uuid(设备编号)进行keyby，结果导致subtask的state大小差异一倍，两种方法，第一，两阶段聚合；第二，重新设置并行度，改变KeyGroup的分布  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1bGVpZGFyZW4_size_16_color_FFFFFF_t_70_pic_center 1]

###  ###

### 第三种 频繁FULL GC【减少key数量；增大TM内存】 ###

当StateSize达到200M以上，Async的时间会超过1min。  
这种情况特别少见，因为RocksDb State的异步阶段做的事情主要是将本地KV数据库里的增量State写到HDFS上，如果flink配置了增量chekcPoint是不太可能出现单个作业异步处理特别慢的现象。因此猜测是由于TM出现频繁FGC，导致线程根本没有足够的时间片去处理。  
结果也确实如此，jstat -gcutil pid 1s，发现每4秒一个fgc。  
dump分析

jmap -dump:format=b,file=jconsole.dump PID  
./ParseHeapDump.sh jconsole.dump org.eclipse.mat.api:suspects  
org.eclipse.mat.api:overview org.eclipse.mat.api:top\_components

还有个有趣的现象是出现FGC时，反压机制会无法生效，在‘BackPressure’界面会一片空白～  
**通过Dump分析，CopyOnWriteStateTable/CopyOnWriteStateMap占用绝大多数堆内存，也就是flink内部用于存储keyedState，CopyOnWriteStateTable 中保存多个 KeyGroup 的状态，每个 KeyGroup 对应一个 CopyOnWriteStateMap。  
解决方法，keyby的key过多，要么减少key的数量，要么加大TM的内存。**

如上，key的Selector定义中有日期，那么就导致key的数量会按天暴涨，也解释了为什么CopyOnWriteMapState对象会这么多了，因为即使KeyedProcessFunction中设置了StateTtl，State会过期，但是Key不会过期。  
分享一篇好文：https://blog.csdn.net/weixin\_44904816/article/details/107995059

### 第四种 出现反压 ###

还有一种情况是当一个作业出现反压时，也会导致超时，表现的形式就是 AcknowledgeTime 都无法拿到，或者 E2E 时间很长，等反压降才去就好了，

参考：

[https://blog.csdn.net/wuleidaren/article/details/108514357][https_blog.csdn.net_wuleidaren_article_details_108514357]

### 解决方法5: 可以使用rocksdb，做增量的checkpoint ###

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1bGVpZGFyZW4_size_16_color_FFFFFF_t_70_pic_center]: /images/20210923/a946505336ee41a8914339d858950359.png
[https_programskills.blog.csdn.net_article_details_115641985]: https://programskills.blog.csdn.net/article/details/115641985
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1bGVpZGFyZW4_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20210923/2b77da0781464bc895d5c07241766cdb.png
[https_blog.csdn.net_wuleidaren_article_details_108514357]: https://blog.csdn.net/wuleidaren/article/details/108514357