我对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 达到记录。
现在我认识到这是一个微不足道的例子，但不理解这一点会使理解更复杂的流变得困难。
2条答案
按热度按时间sycxhyv71#
如果使用BoundedAutoFordernessTimestampExtractor，则在出现新事件之前不会输出最后一次计算。如果我们在水印中使用systemtime，它可以工作，但是当您重新运行带有嵌入时间戳（过去事件）的消息时，它不会计算这些时间戳。
klsxnrf12#
你的理解基本上是正确的，但这里还有一些事情需要考虑。
首先，你用了
assignAscendingTimestamps()，它只能在事件流完全有序（按时间戳）时使用，这里不是这样。运行此应用程序时应看到以下警告：另一个因素是
AscendingTimestampExtractor不会为每个传递的流元素更新当前水印。这是一个周期性水印生成器的示例，它将注入Watermark每n毫秒进入一次流，其中n由ExecutionConfig.setAutoWatermarkInterval(...)，默认为200毫秒。这就是事件4如何潜入第一个窗口。要获得预期的结果，可以实现一个标点水印生成器，该生成器配置为为为每个事件生成水印：
你可以这样使用：
现在，您的示例生成以下结果：
事件#4已被删除，因为时间已晚。这可以通过放松水印生成器来调整，以适应一些无序的情况。例如。，
然后产生这些结果：
或者您可以配置窗口以允许延迟事件
这会导致第一扇Windows着火两次：
请注意，由于处理水印会带来一些开销，因此通常不希望以这种方式使用带标点的水印（每个事件都有一个水印）。对于大多数应用，周期性水印基于
BoundedOutOfOrdernessTimestampExtractor是更好的选择。