窗口触发器不返回最新的结果

Question

我正在尝试测量具有窗口操作的 Flink 应用程序的延迟，如下所示：

SingleOutputStreamOperator<String> branch = stream
                .getSideOutput(outputTag2)
                .keyBy(MetricObject::getRootAssetId)
                .window(TumblingEventTimeWindows.of(Time.seconds(60)))
                .trigger(ContinuousEventTimeTrigger.of(Time.seconds(15)))
                .aggregate(new CountDistinctAggregate(), new CountDistinctProcess())
                .name("windowed-count-distinct")
                .uid("windowed-count-distinct")
                .map((value)->String.valueOf(value.getTimestamp().toEpochMilli()))
                .name("send-timestamp");

我正在考虑事件时间并提取时间戳，我使用这个水印策略：

                        .<SingleRecord>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                        .withTimestampAssigner((event, timestamp) -> event.getTimestamp().toEpochMilli()))

聚合函数将特定对象保存为累加器，其中还包含提取的时间戳;这些时间戳写在 kafka 主题中。问题是返回的时间戳如下：

1639651859988
1639651890163
1639651904900
1639651919728
1639651919728
1639651949973
1639651965085
1639651979870

返回的时间戳并不像我预期的那样间隔相等，第四个和第五个是相等的，但它们的返回间隔为15秒，这是不可能的，因为应用程序记录的输入每秒（每秒10个）连续生成。在其他测试中，我也遇到了更糟糕的情况，比如：

1639651979870
1639651992771
1639651992771
1639651992771
1639651992771
1639652189791
1639652205001
1639652219876

奇怪的事实是，当我使用一个简单的翻滚窗口而不触发时：

 .window(TumblingEventTimeWindows.of(Time.seconds(15)))

返回的时间戳按预期等距分布:

1639652429766
1639652444930
1639652459900
1639652474609
1639652489746
1639652504862
1639652519734
1639652534847

我真的不明白这是问题所在，似乎聚合函数中的累加器没有正确升级。

Answer 1

我认为您可以检查Flink streaming的输入数据，以验证结果是否符合预期。

对于第一个流式传输，聚合操作在60秒窗口的数据集中运行4次（每15秒）。不确定您在聚合中的逻辑。例如。假设我们有一个3秒的窗口，触发器是每1秒一次。运算符是在窗口中获取max元素。还假设输入每1秒生成一次。如果输入是1、3、2、...，那么我们将看到Flink的输出，如1、3、3...，因为第一个窗口在其窗格中有[1、3、2]，对于每个获取max元素的触发器，结果将是1、3、3。

对于第二个流式作业，每个窗口都有一个触发时间，因此，例如，使用上面的输入作为例子，如果窗口是1秒，我们将得到1、3、2、...