Apache Flink滚动窗口数据拆分问题排查咨询

]

}

后续将消息转换为包含listId和产品的元组数据流，最终通过`KeyBy`和10秒滚动处理时间窗口，按listId和fatherId分组后转换为指定格式字符串。测试时发送5条各包含128000条数据的列表，预期得到5条结果字符串，但偶尔会出现6条——其中一条原始消息对应的结果被拆分，本该是单条字符串。

处理流程的核心代码如下：
```java
DataStream<Result> sourceNegotiation = listNegotiationProducts
                .flatMap(new FlatMapFunction<ListNegotiationProduct, Tuple2<UUID, NegotiationProduct>>() {
                    @Override
                    public void flatMap(ListNegotiationProduct listNegotiationProduct, Collector<Tuple2<UUID, NegotiationProduct>> out) throws Exception {
                        listNegotiationProduct.getProducts().forEach(lnp -> {
                            Tuple2<UUID, NegotiationProduct> response = new Tuple2<>(listNegotiationProduct.getTransactionId(), lnp);
                            out.collect(response);
                        });
                    }
                })
                .keyBy(new KeySelector<Tuple2<UUID, NegotiationProduct>, Tuple2<UUID, Integer>>() {
                    @Override
                    public Tuple2<UUID, Integer> getKey(Tuple2<UUID, NegotiationProduct> value) throws Exception {
                        return Tuple2.of(value.f0, value.f1.getNegotiationId());
                    }
                })
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .allowedLateness(Time.seconds(1))
                .apply(new WindowFunction<Tuple2<UUID, NegotiationProduct>, Tuple2<UUID, Negotiation>, Tuple2<UUID, Integer>, TimeWindow>() {
                    @Override
                    public void apply(Tuple2<UUID, Integer> uuidIntegerTuple2, TimeWindow window, Iterable<Tuple2<UUID, NegotiationProduct>> iterable, Collector<Tuple2<UUID, Negotiation>> collector) throws Exception {
                        Negotiation negotiation = new Negotiation();
                        Tuple2<UUID, Negotiation> response = new Tuple2<>();

                        List<Product> productList = new ArrayList<>();

                        iterable.iterator().forEachRemaining(negotiationProduct -> {

                            negotiation.setNegotiationId(negotiationProduct.f1.getNegotiationId());
                            response.setField(negotiationProduct.f0, 0);

                            List<String> observationList = new ArrayList<>();

                            observationList.add(negotiationProduct.f1.getObservation());

                            productList.add(Product
                                    .builder()
                                    .productGtin(negotiationProduct.f1.getProductGtin())
                                    .state(negotiationProduct.f1.getState())
                                    .observation(observationList)
                                    .retailerCode(negotiationProduct.f1.getRetailerCode()).build());
                        });

                        negotiation.setNegotiationProgressProducts(productList);

                        response.setField(negotiation, 1);
                        collector.collect(response);
                    }
                })
                .keyBy(t -> t.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .allowedLateness(Time.seconds(1))
                .apply(new WindowFunction<Tuple2<UUID, Negotiation>, Result, UUID, TimeWindow>() {
                    @Override
                    public void apply(UUID uuid, TimeWindow window, Iterable<Tuple2<UUID, Negotiation>> iterable, Collector<Result> collector) throws Exception {
                        List<Negotiation> negotiations = new ArrayList<>();
                        iterable.iterator().forEachRemaining(n -> {
                            negotiations.add(n.f1);
                        });
                        collector.collect(BuildResult.build(new Payload(negotiations), uuid));
                    }
                })
                .returns(Result.class);

• 窗口迟到数据触发重复计算：
  流程中两次使用10秒滚动处理时间窗口，且都设置了1秒允许迟到时间。当单条原始消息展开的128000个元组因处理延迟，部分落在第一个窗口的时间范围内，部分在窗口结束后1秒内到达，会触发第一个窗口二次计算，输出多份对应同一个UUID的Tuple2<UUID, Negotiation>。后续第二个按UUID分组的窗口会在不同周期收集到这些数据，最终输出多条结果字符串。

• 并行度导致数据分散处理：
  若Flink并行度较高，同一个原始消息的元组可能被分配到不同并行子任务处理。不同子任务的处理速度差异会导致部分元组错过第一个窗口触发时间，进入迟到数据流程，进而让后续窗口收集到拆分的数据。

• 处理时间的不确定性：
  滚动窗口基于处理时间触发，而处理时间受系统负载、资源分配影响。当系统繁忙时，部分元组的处理时间被拉长，可能跨越窗口时间边界，导致同一个UUID的数据被拆分到多个窗口处理，最终输出多条结果。

• 检查输出结果的UUID，确认是否存在同一个UUID对应多条结果的情况，验证是否由窗口重复触发导致。
• 调整窗口参数：延长第一个窗口时长，或缩短允许迟到时间，减少跨窗口的迟到数据。
• 优化并行度设置：通过自定义Partitioner或调整KeyBy逻辑，确保同一个原始消息的元组被分配到同一个子任务处理，避免并行处理带来的时间差。
• 考虑使用事件时间窗口替代处理时间窗口，基于消息中的时间戳对齐窗口，降低处理时间波动的影响。

内容的提问来源于stack exchange，提问作者sergio garcia