Apache Flink中带有TumblingWindow的水印

Question

我试图了解Apache FLink中Windows和Watermark生成之间的依赖关系，我在下面的示例中出现错误：

   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，4 C，8，其中C是类别，5是时间戳。

每当我发送事件时，数据流都会打印，但不会使用窗口打印这些事件（打印（“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秒，因此您的应用程序必须运行很长时间才能生成第一个水印。（我想不出设置这么大的水印间隔有什么帮助。）

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