AggregateFunction getResult() 在事件聚合后未被调用

Question

尝试实现一个 Flink 作业来读取 Kafka 流并聚合会话，由于某种原因 getResult(( 没有被调用。我看到 createAccumulator(( 和 add(( 被调用，我希望 getResult(( 也被调用，以便我可以在目标中接收聚合消息。

source.keyBy(new KeySelector<GenericRecord, String>() {
@Override
public String getKey(GenericRecord record) {
return record.get("id").toString();
}})
.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<GenericRecord>() {
private static final long serialVersionUID = -4834111073247835189L;
private final long maxTimeLag = 300000L;
@Nullable
@Override
public Watermark checkAndGetNextWatermark(GenericRecord lastElement, long extractedTimestamp) {
return new Watermark(extractedTimestamp - maxTimeLag);
}
@Override
public long extractTimestamp(GenericRecord element, long previousElementTimestamp) {
long ts = 1000 * (long)element.get(("timestamp"));
return (ts);
}
})
.map(new ReduceAttributesMap())
.keyBy(new KeySelector<Tuple2<String, String>, String>() {
@Override
public String getKey(Tuple2<String, String> e) {
return e.f0;
}
})
.window(EventTimeSessionWindows.withGap(Time.minutes(5)))
.aggregate(new EventAggregation())
.addSink(...)

可能是什么问题？我是否配置错误？感谢您的帮助！

Answer 1

仅在窗口完成时调用AggregateFunction#getResult()。在您的情况下，仅当 5 分钟后特定键没有事件时，才会发出窗口。你能在你的数据中确认这个案例确实发生了吗？

您可以尝试减少会话窗口的间隔时间，以便更轻松地查看它。此外，您的水印分配器看起来很可疑。您可能想使用BoundedOutOfOrdernessTimestampExtractor.最后，您可以仔细检查您的时间提取是否按预期工作吗？时间戳是否存储为自 1970 年以来的秒？