Apache Flink：窗口还原函数永远不会执行

Question

下面是代码片段，我在其中使用了基于翻滚事件时间的窗口

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函数可以正常工作。下面是时间戳提取器的代码。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，因为我的窗口也是以秒为单位定义的。< br >当我使用返回分钟时间戳的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。它的工作原理是跟踪到目前为止在数据流中看到的最大时间戳，并从该最大时间戳（而不是系统时钟）中减去延迟。源代码，以防您好奇。