Apache Flink：Windowed ReduceFunction 永远不会执行

Question

下面是代码 sn-p，我使用的是基于 Tumbling EventTime 的窗口

DataStream<OHLC> ohlcStream = stockStream.assignTimestampsAndWatermarks(new TimestampExtractor()).map(new mapStockToOhlc()).keyBy((KeySelector<OHLC, Long>) o -> o.getMinuteKey())
        .timeWindow(Time.seconds(60))
        .reduce(new myAggFunction());

不幸的是，它看起来从不执行 reduce 函数。如果使用上面没有窗口的代码，reduce 函数可以正常工作。下面是 TimestampExtractor 的代码。 30秒的水印延迟只是一个测试值，但一分钟的翻滚窗口是m

    public static class TimestampExtractor implements AssignerWithPeriodicWatermarks<StockTrade> {
    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(System.currentTimeMillis() - 30000);
    }

    @Override
    public long extractTimestamp(StockTrade stockTrade, long l) {
        BigDecimal bd = new BigDecimal(stockTrade.getTime());
        // bd contains miliseconds timestamp 1498658629.036
        return bd.longValue();
    }
}

bd.longValue() 返回秒时间戳 1498658629，因为我的窗口也是以秒为单位定义的。
当我使用返回分钟时间戳的 bd.longValue()/60 时，会调用 reduce 函数。我的输出文件包含每个reduce操作的所有记录

{time=1498717692.000, minuteTime=24978628, n=1, open=2248.0}
{time=1498717692.000, minuteTime=24978628, n=2, open=2248.0}
...
{time=1498717692.000, minuteTime=24978628, n=8, open=2248.0}

那么，谁能给我解释一下，发生了什么？非常感谢。

Answer 1

通常，水印应与数据中的时间戳相关，而不应基于系统时钟。使用事件时间的一大好处是，同一个应用程序可用于重新处理历史数据或处理当前数据，但如果您将时间戳与系统时钟进行比较，就像您在此处所做的那样，这是不可能的。

水印可以被认为是所有时间戳小于水印的数据已经到达的声明。或者换句话说，任何时间戳小于当前水印的数据都将被认为是迟到的。我的猜测是您没有看到任何结果，因为您的水印导致您的所有数据都被认为是迟到的，并且窗口运算符正在删除所有这些迟到的数据。

我建议您改用BoundedOutOfOrdernessTimestampExtractor。它的工作原理是跟踪迄今为止在数据流中看到的最大时间戳，并从该最大时间戳中减去延迟，而不是系统时钟。 source code，如果你好奇的话。