并行度改变引发的血案

2022-04-18 13:24:13 浏览数 (1)

收到电话告警kafka消费堆积,就去查看任务健康状态,查找了以下几点:

1.查看背压情况,web页面绿码通行

2.查看任务GC情况,正常

3.查看数据写入情况,外部数据库mysql指标正常

4.查看checkpoint情况,几十毫秒完成

5.topic 生产消费速度,震惊~ 生产速度double了

至此可以确认消费能力不足导致,那就使用增加资源大法,调大任务并行度,看似一起都非常完美,

一顿操作调大并行度,重启任务,wath ? 任务咋报错了呢,查看错误日志,大概说的是写入mysql唯一键冲突,内心万马鹏腾,老老实实翻开代码看一下写入逻辑:做了一个窗口聚合的业务逻辑将窗口聚合的结果写入mysql,延时数据也不能丢弃,因此加了一个sideOutput 获取延时数据,为了减轻写入的压力,会将需要输出的数据缓存在状态中,做了定时定量输出,对于窗口结果数据是按照keyBy的,key与Mysql唯一键对应,那么对于全局来说key windowTime 肯定是唯一的,对于延时数据来说拿到的是一条条的明细数据,没有经过窗口的处理的,因此在写入mysql之前会手动做窗口分配合并然后输出。看到这里,已经发现问题的根源,缓存数据状态使用的是operator-list 类型,改变任务并行度,会导致list数据被重新分配到不同的task中,对于延时的数据很有可能就会出现在不同的task出现属于同一个key windowTime的数据,那么在写入的时候就会出现一个先写成功,一个后写就失败了。

解决方案:

1.首先并行度不做改变,在initializeState 方法中,将获取的状态数据直接刷写到mysql中

2.延时数据在写入到缓存时,做一次窗口分配、合并操作,保证延时缓存中的数据key windowTime是唯一的

3.最后重新调整任务并行度

至此bug解决完成,做事还是不能太嚣张啊~

回顾一下任务并行度改变对状态产生的影响:

1.对于keyed state , flink 在状态恢复的时候会按照key group自动完成状态的重新分配

2.对于operator state来说,根据使用的不同类型状态会分为以下几种情况:list state 会以轮序的方式重新分配,例如kafka offset; union state 所有的task都会有一份全量的状态信息;broadcast state 只会被广播一份数据

0 人点赞