雅安公司做网站,网站备案用的幕布,企业网站下载,注册网站用的信用卡文章目录 前言一、现象二、结论三、解决 前言
窗口没有关闭计算的问题#xff0c;一直困扰了很久#xff0c;经过多次验证#xff0c;确定了问题的根源。 一、现象
Flink使用了window#xff0c;同时使用了watermark #xff0c;并且还设置了较高的并行度。生产是设置了… 文章目录 前言一、现象二、结论三、解决 前言
窗口没有关闭计算的问题一直困扰了很久经过多次验证确定了问题的根源。 一、现象
Flink使用了window同时使用了watermark 并且还设置了较高的并行度。生产是设置了300的并行度并且接入了几十个topic 这个地方划重点后面会提到。结果就是窗口没有关闭进行计算。于是我查阅的相关文档得到的答案是因为配置的源并行度大于topic的分区数而导致。这个答案只能说很接近而且我最开始也觉得很有道理。 解释一下watermark window的原理 可以看到前面三个窗口里面都有数据窗口触发计算的其中一个必要条件是最新的数据没过最低的水位线就进行计算认为不会再有乱序的数据进来了。但是从图中我们可以看到其中一个窗口一个数据都没有就会导致拿不到所有窗口的最低水位线。因此也就无法触发计算。 为了验证这一法则 我在测试环境配置了一个并行度为10的程序topic只有一个分区启动任务的时候我信誓旦旦地保证这不可能关闭窗口进行计算然而现实狠狠打了我一巴掌窗口结果算出来了。虽然只是三言两语实际上我做了很多尝试只是其他的实验不重要都是证明我是错的
于是通过比较的方法想到和生产的情况不同就在于生产消费了几十个topic而我的测试只有一个topic于是我再次坚信问题一定就在这了。
我直接在idea进行测试
我配置了两个topic并且在一开始只往第一个topic中写数据而第二个topic不写数据
很好跑了一整个中午一次窗口聚合计算都没有。
此时进行最后一步验证就是往第二个topic写数据。
我在这个时间往第二个topic发了数据
collectTime:1697693856606为了让大家看清楚现象我把日志和截图都给出来
2023-10-19 13:37:32.699 [Legacy Source Thread - Source: Custom Source - Flat Map - (Flat Map - Flat Map - Sink: Unnamed, Timestamps/Watermarks - (Flat Map, Flat Map, Flat Map)) (10/16)#0] INFO c.a.c.d.risk.domain.function.IndicatrixMapFunction - 【通过】滑动窗口前置数据处理
2023-10-19 13:37:32.805 [Window(TumblingEventTimeWindows(60000), EventTimeTrigger, CountAverageFunction, LogResultWindowFunction) (13/16)#0] INFO com.ai.cass.dc.risk.re.idxSend.IdxSend - 聚合时存储指标结果,calcTypeCode:FrequencyOccurStttc key:ff83d41c-335f-405d-88e7-f5285aecdcf5a1123 Value:8
2023-10-19 13:37:32.805 [Window(TumblingEventTimeWindows(60000), EventTimeTrigger, CountAverageFunction, LogResultWindowFunction) (13/16)#0] INFO com.ai.cass.dc.risk.re.idxSend.IdxSend - 聚合时存储指标结果,calcTypeCode:FrequencyOccurStttc key:ff83d41c-335f-405d-88e7-f5285aecdcf5a1123 Value:27
2023-10-19 13:37:32.805 [Window(TumblingEventTimeWindows(60000), EventTimeTrigger, CountAverageFunction, LogResultWindowFunction) (13/16)#0] INFO com.ai.cass.dc.risk.re.idxSend.IdxSend - 聚合时存储指标结果,calcTypeCode:FrequencyOccurStttc key:ff83d41c-335f-405d-88e7-f5285aecdcf5a1123 Value:28
2023-10-19 13:37:32.805 [Window(TumblingEventTimeWindows(60000), EventTimeTrigger, CountAverageFunction, LogResultWindowFunction) (13/16)#0] INFO com.ai.cass.dc.risk.re.idxSend.IdxSend - 聚合时存储指标结果,calcTypeCode:FrequencyOccurStttc key:ff83d41c-335f-405d-88e7-f5285aecdcf5a1123 Value:17
2023-10-19 13:37:32.805 [Window(TumblingEventTimeWindows(60000), EventTimeTrigger, CountAverageFunction, LogResultWindowFunction) (13/16)#0] INFO com.ai.cass.dc.risk.re.idxSend.IdxSend - 聚合时存储指标结果,calcTypeCode:FrequencyOccurStttc key:ff83d41c-335f-405d-88e7-f5285aecdcf5a1123 Value:20证明就是在这个时间节点上窗口计算处理结果
二、结论
因此我就可以大胆地推断是因为多个topic进行了数据消费其中有个topic数据会进入窗口进行计算但有的窗口又永远不会有数据进入计算这就造成对应的窗口永远没有最低的watermark以致于窗口无法关闭并计算。
三、解决
既然问题找到了那解决办法就随之而生
1、如果可以不使用水印直接关闭水印即可只要消费的数据不会积压并且要求没那么高的话这个方法最简单2、减小并行度到能够使得每个窗口都有数据减小并行度会让不同topic用同一个窗口至于这个数量那还得研究研究了3、把需要到窗口和不到窗口计算的数据进行分流4、也可以把源与后面算子之间采用rebalance的方式传递这样就能够轮询的方式往下传递使得每个window都会有数据这里有一点一定要注意rebalance必须放在watermark之前才可以。
