Flink如何设置初始水印

Question

我正在使用Flink 1.3.2和scala构建一个流媒体应用程序，我的Flink应用程序将监视一个文件夹，并将新文件流到管道中。文件中的每条记录都有一个相关的时间戳。我想使用此时间戳作为事件时间，并使用AssignerWithPeriodicWatermarks构建水印，我的水印生成器如下所示：

class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[Activity] {
    val maxTimeLag = 6 * 3600000L // 6 hours
    override def extractTimestamp(element: Activity, previousElementTimestamp: Long): Long = {
    val format = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssXXX")
        val timestampString = element.getTimestamp
    }
    override def getCurrentWatermark(): Watermark = {
      new Watermark(System.currentTimeMillis() - maxTimeLag)
    }
  }

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.getConfig.setAutoWatermarkInterval(10000L)
val stream = env.readFile(inputformart, path, FileProcessingMode.PROCESS_CONTINUOUSLY, 100)

val activity = stream
      .assignTimestampsAndWatermarks(new TimeLagWatermarkGenerator())
      .map { line =>
        new tuple.Tuple2(line.id, line.count)
      }.keyBy(0).addSink(...)

但是，由于我的文件夹中有一些旧数据，我不想处理它们。旧文件中记录的时间戳是

Answer 1

您所展示的管道中没有关心时间的操作符——没有窗口，没有ProcessFunction计时器——因此每个流元素都将畅通无阻地通过并被处理。如果您的目标是跳过延迟的元素，则需要引入（以某种方式）将事件时间戳与当前水印进行比较的内容。

您可以通过在keyBy和接收器之间引入一个步骤来做到这一点，如下所示：

...
.keyBy(0)
.process(new DropLateEvents())
.addSink(...)

public static class DropLateEvents extends ProcessFunction<...> {
    @Override
    public void processElement(... event, Context context, Collector<...> out) throws Exception {
        TimerService timerService = context.timerService();
        if (context.timestamp() > timerService.currentWatermark()) {
           out.collect(event);
        }
    }
}

完成此操作后，您关于初始水印的问题就变得相关了。对于周期性水印，初始水印是Long.MIN_VALUE，因此在发出第一个水印之前不会考虑延迟，这将在操作10秒后发生（给定您如何设置自动水印间隔）。

如果您想更详细地了解如何生成周期性水印，请参阅此处的相关代码。

如果要避免在前10秒内处理延迟元素，可以完全忘记使用事件时间和水印，只需修改上面显示的processElement方法，将事件时间戳与系统进行比较。currentTimeMillis（）-maxTimeLag，而不是当前水印。另一种解决方案是使用带标点的水印，并在第一个事件中发出水印。

或者更简单地说，您可以在平面图或过滤器中检测并删除延迟事件，因为您定义的是相对于系统的延迟。currentTimeMillis（）而不是水印。