Apache Flink中带有翻滚窗口的水印

Question

我试图理解Apache FLink中Windows和水印生成之间的依赖关系，我在下面的示例中遇到了一个错误：

public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setAutoWatermarkInterval(10000);
FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<String>("watermarkFlink", new SimpleStringSchema(), props);
DataStream<String> orderStream = env.addSource(kafkaSource);
DataStream<Order> dataStream = orderStream.map(str -> {
String[] order = str.split(",");
return new Order(order[0], Long.parseLong(order[1]), null);
});
WatermarkStrategy<Order> orderWatermarkStrategy = CustomWatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(1))
.withTimestampAssigner((element, timestamp) ->
element.getTimestamp()
);
dataStream
.assignTimestampsAndWatermarks(orderWatermarkStrategy)
.map(new OrderKeyValue())
.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public String getKey(Tuple2<String, Integer> src) throws Exception {
return src.f0;
}
})
.window(SlidingEventTimeWindows.of(Time.seconds(20), Time.seconds(5)))
.sum(1)
.print("Windows");
dataStream.print("Data");
env.execute();
}

public static class OrderKeyValue implements MapFunction<Order, Tuple2<String, Integer>> {
@Override
public Tuple2<String, Integer> map(Order order) {
return new Tuple2<>(order.getCategory(), 1);
}
}

这里的时间戳很长，我们可以从Kafka源中检索到，它应该是：a，4C，8，其中C是类别，5是时间戳。

每当我发送一个事件时，数据流都在打印，而不是用窗口打印(print("Windows"((。此外，例如，如果我收到一个事件A，12，然后我生成了一个水印(在10秒内(，那么我有第一个窗口关闭后的C，2，它会在窗口中处理还是被忽略？

Answer 1

Flink文档中有一个教程应该有助于澄清这些概念：https://nightlies.apache.org/flink/flink-docs-stable/docs/learn-flink/streaming_analytics/

但总结一下情况：

• 如果您有一个类似(A，4((C，8((A，12(的事件流，那么这些整数将被解释为毫秒。

• 您的第一个窗口将等待20000的水印，然后才会被触发。

• 要生成这么大的水印，您需要一个时间戳至少为21000的事件(因为有界无序设置为1秒(。

• 由于您已将自动水印间隔配置为10秒，因此您的应用程序必须运行该时间才能生成第一个水印。(我想不出在任何情况下，将水印间隔设置得这么大都有帮助。(

• 如果事件在其窗口关闭后到达，则它将被忽略(默认情况下(。您可以配置允许的延迟来安排延迟事件以触发额外的窗口触发。