使用Flink和基于事件时间的流计算平均值

Question

我想用基于历史事件的流计算Flink中基于窗口的平均值（或我定义的任何其他函数），因此流必须是事件时间（而不是基于处理时间）：

val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

我已经了解了如何在摄入时添加时间戳：

ctx.collectWithTimestamp(Datapoint(instrument, bid, ask), time.getMillis)

但是当我进行计算（应用函数）时，当我只是以与没有EventTime时相同的方式进行计算时，它就不起作用了。我读过一些关于我必须设置的水印的东西：

val avg = stream
  .keyBy("instrument")
  .timeWindow(Time.seconds(10))
  .apply((key: Tuple, window: TimeWindow, values: Iterable[Datapoint], out: Collector[Datapoint])=>{
    val avg = values.map(_.val).sum / values.size
    val dp = Datapoint(key.getField[String](0), avg)
    out.collect(dp)
  })

avg.print()
env.execute()

有没有人举一个简单的Scala例子？

尊敬的安德烈亚斯

Answer 1

水印是一种有效的时间戳，它断言具有较早时间戳的所有事件（可能）已经到达。基于事件时间的窗口依赖于水印来知道窗口何时完成。到目前为止，最常见的水印策略是假设事件到达时有一定的延迟。

如果要在数据源中发出水印（在摄取期间），请参阅带有时间戳和水印的源函数，但它将像

ctx.emitWatermark(new Watermark(datapoint.getWatermarkTime))

另一方面，如果要在源代码之外处理此问题，请参阅时间戳赋值器/水印生成器和允许固定延迟量的赋值器。您可以简单地执行以下操作：

stream
  .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Datapoint](Time.seconds(10))( _.getTimestamp ))
  .keyBy("instrument")
  ...

我链接到的文档中有更详细的Scala示例。