升级 Flink 已弃用的函数调用

Question

我目前正在尝试升级应用于数据流的方法调用assignTimestampsAndWatermarks。数据流如下所示：

DataStream<Auction> auctions = env.addSource(new AuctionSourceFunction(auctionSrcRates))
        .name("Custom Source")
        .setParallelism(params.getInt("p-auction-source", 1))
        .assignTimestampsAndWatermarks(new AuctionTimestampAssigner());

AssignerWithPeriodicWatermark 如下所示：

private static final class AuctionTimestampAssigner implements AssignerWithPeriodicWatermarks<Auction> {
        private long maxTimestamp = Long.MIN_VALUE;

        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(maxTimestamp);
        }

        @Override
        public long extractTimestamp(Auction element, long previousElementTimestamp) {
            maxTimestamp = Math.max(maxTimestamp, element.dateTime);
            return element.dateTime;
        }
    }

我需要采取哪些步骤才能从弃用的调用升级到当前的最佳做法？谢谢。

Answer 1

您的水印生成器假定事件按时间戳按顺序排列，或者至少接受任何无序事件都会迟到。这相当于

assignTimestampsAndWatermarks(
    WatermarkStrategy
      .<Auction>forMonotonousTimestamps()
      .withTimestampAssigner((event, timestamp) -> event.dateTime))