Flink水印根本没有前进？卡在-9223372036854775808

Question

我遇到了与Flink EventTime处理水印类似的问题-9223372036854725808。然而，建议的解决方案（设置并行度和禁用检查点）没有任何效果。在本例中，我只是将1000个事件以1秒的间隔流式传输，然后将事件时间戳与ctx进行比较。timerService（）。currentWatermark（）

>>> v=(61538659200000,0), watermark=-9223372036854775808
>>> v=(61538659201000,1), watermark=-9223372036854775808
>>> v=(61538660198000,998), watermark=-9223372036854775808
>>> v=(61538660199000,999), watermark=-9223372036854775808


public void watermarks()
    throws Exception
{
    final var env = StreamExecutionEnvironment.createLocalEnvironment();
    env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    env.setMaxParallelism(1);

    final long startMs = new Date(2020, 1, 1).getTime();
    final var events = new ArrayList<Tuple2<Long, Integer>>();
    for (var ii = 0; ii < 1000; ++ii ) {
        events.add(new Tuple2<Long, Integer>(startMs + ii * 1000, ii));
    }

    env.fromCollection(events)
        .assignTimestampsAndWatermarks(
            WatermarkStrategy.<Tuple2<Long, Integer>>forMonotonousTimestamps()
                .withTimestampAssigner((event, ts) -> event.f0))
        .setParallelism(1)
        .keyBy(row -> row.f1 % 2)
        .process(new ProcessFunction<Tuple2<Long, Integer>, String>()
        {
            @Override
            public void processElement(
                final Tuple2<Long, Integer> value,
                final Context ctx,
                final Collector<String> out)
                throws Exception
            {
                out.collect("v=" + value + ", watermark=" + ctx.timerService().currentWatermark());
            }
        })
        .setParallelism(1)
        .print()
        .setParallelism(1);
    final var result = env.execute();
    System.out.println(result);
}

Answer 1

forMonotonousTimestamps是一个周期性水印生成器，仅在由计时器触发时生成水印。默认情况下，此计时器每200毫秒触发一次（这是autoWatermark Interval）。您的作业运行时间不足以让此计时器触发。

有界源确实会生成一个水印，其时间戳设置为MAX_WATERMARK当它们到达输入末尾时——就在关闭作业之前。您在作业的输出中看不到这个水印，因为它后面没有事件。

如果您想为每个事件生成水印，您可以实现自定义水印策略，在Watermark Generator（docs）的onEvent方法中发出水印。这在生产中通常是个坏主意，因为您会在这些额外的水印上浪费CPU周期和网络带宽，但有时对于测试来说这很有帮助。