文章目录
- 前言
- 一、现象
- 二、结论
- 三、解决
前言
窗口没有关闭计算的问题，一直困扰了很久，经过多次验证，确定了问题的根源。
一、现象
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之前才可以。
