Flink 水印和触发器 - 事件时间未丢弃的后期元素?

Question

我对 Flink 在事件时间加水印时如何处理后期元素感到有些困惑。

我的理解是，当 Flink 读取数据流时，当看到任何事件时间比当前水印大的数据时，水印时间就会取得进展。然后，任何覆盖时间严格小于水印的窗口都会被触发驱逐(假设没有延迟允许。

但是，以这个最小的例子为例：

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector
import org.apache.log4j.{Level, Logger}
object EventTimeExample {
Logger.getLogger("org").setLevel(Level.OFF)
Logger.getLogger("akka").setLevel(Level.OFF)
case class ExampleType(time: Long, value: Long)
def main(args: Array[String]) {
// Set up environment
val env = StreamExecutionEnvironment.createLocalEnvironment(1)
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
// Example S3 path
val simple = env.fromCollection(Seq(
ExampleType(1525132800000L, 1),
ExampleType(1525132800000L, 2) ,
ExampleType(1525132920000L, 3),
ExampleType(1525132800000L, 4)
))
.assignAscendingTimestamps(_.time)
val windows = simple
.windowAll(TumblingEventTimeWindows.of(Time.seconds(60)))
.apply{
(window, iter, collector: Collector[(Long, Long, String)]) => {
collector.collect(window.getStart, window.getEnd, iter.map(_.value).toString())
}
}
windows.print
env.execute("TimeStampExample")
}
}

运行此程序的结果是：

(1525132800000,1525132860000,List(1, 2, 4))
(1525132920000,1525132980000,List(3))

但是，如果我的理解是正确的，则4不应包含在此处的第一个窗口中，因为当达到记录的值时，应更新水印时间3。

现在我认识到这是一个微不足道的例子，但不理解这一点会使理解更复杂的流程变得困难。

Answer 1

你的理解基本上是正确的，但这里还有一些事情需要考虑。

首先，您已经使用了assignAscendingTimestamps()，它只能在事件流完全有序(按时间戳(时使用，而这里的情况并非如此。运行此应用程序时，应会看到以下警告：

WARN  org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor  - Timestamp monotony violated: 1525132800000 < 1525132920000

这里起作用的另一个因素是AscendingTimestampExtractor不会为每个传递的流元素更新当前水印。这是一个周期性水印生成器的示例，它将每 n 毫秒向流注入一个Watermark，其中n由ExecutionConfig.setAutoWatermarkInterval(...)定义，默认为 200 毫秒。这就是事件 #4 潜入第一个窗口的方式。

要获得预期的结果，您可以实现一个标点水印生成器，配置为为每个事件生成水印：

class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[ExampleType] {
override def extractTimestamp(element: ExampleType, previousElementTimestamp: Long): Long = {
element.time
}
override def checkAndGetNextWatermark(lastElement: ExampleType, extractedTimestamp: Long): Watermark = {
new Watermark(extractedTimestamp)
}
}

然后你会像这样使用：

val simple = env.fromCollection(Seq(
ExampleType(1525132800000L, 1),
ExampleType(1525132800000L, 2) ,
ExampleType(1525132920000L, 3),
ExampleType(1525132800000L, 4)
))
.assignTimestampsAndWatermarks(new PunctuatedAssigner)

现在，您的示例生成以下结果：

(1525132800000,1525132860000,List(1, 2))
(1525132920000,1525132980000,List(3))

事件 #4 已被丢弃，因为它已晚。这可以通过放宽水印生成器来调整，以适应一定程度的无序。例如，

override def checkAndGetNextWatermark(lastElement: ExampleType, extractedTimestamp: Long): Watermark = {
new Watermark(extractedTimestamp - 200000)
}

然后产生以下结果：

(1525132800000,1525132860000,List(1, 2, 4))
(1525132920000,1525132980000,List(3))

或者，您可以将窗口配置为允许延迟事件

val windows = simple
.windowAll(TumblingEventTimeWindows.of(Time.seconds(60)))
.allowedLateness(Time.seconds(200))
...

然后导致第一个窗口触发两次：

(1525132800000,1525132860000,List(1, 2))
(1525132800000,1525132860000,List(1, 2, 4))
(1525132920000,1525132980000,List(3))

请注意，由于处理水印会产生一些开销，因此您通常不希望以这种方式使用标点水印(每个事件都有一个水印(。对于大多数应用程序，基于BoundedOutOfOrdernessTimestampExtractor的定期水印是更好的选择。

Answer 2

如果使用 BoundedOutOfOrdernessTimestampExtractor，则在出现新事件之前不会输出最后一个计算。如果我们在水印中使用 SystemTime，它可以工作，但是当您重新运行带有嵌入时间戳的消息(过去的事件(时，它不会计算这些消息。