apache-flink - 处理时间窗口不适用于 Apache Flink 中的有限数据源

Question

我正在尝试将一个非常简单的窗口函数应用于 Apache Flink 中的有限数据流(本地，无集群)。这是示例:

val env = StreamExecutionEnvironment.getExecutionEnvironment
env
  .fromCollection(List("a", "b", "c", "d", "e"))

  .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1)))
  .trigger(ProcessingTimeTrigger.create)
  .process(new ProcessAllWindowFunction[String, String, TimeWindow] {
    override def process(context: Context, elements: Iterable[String], out: Collector[String]): Unit = {
      out.collect(elements.toList.sorted.toString())
    }
  })

  .print()

env.execute()

在这里，我尝试在一秒钟内将所有到达窗口的元素分组，然后只打印这些组。

我假设所有元素都将在不到一秒的时间内生成并进入一个窗口，因此 print() 中将有一个传入元素。但是，当我运行它时，什么也没有打印。

如果我删除所有的窗口内容，比如

val env = StreamExecutionEnvironment.getExecutionEnvironment
env
  .fromCollection(List("a", "b", "c", "d", "e"))
  .print()

我看到运行后打印的元素。我也用文件源试过这个，没有区别。

我机器上的默认并行度是 6。如果我试验并行度和延迟级别，就像这样

val env = StreamExecutionEnvironment.createLocalEnvironment(2)
env
  .fromCollection(List("a", "b", "c", "d", "e"))
  .map { x => Thread.sleep(1500); x }

我能够将一些(不是全部)元素分组，然后打印出来。

我的第一个假设是源完成的速度比 1 秒快得多，并且任务在窗口的计时器触发之前关闭。调试显示the timer setting line in ProcessingTimeTrigger到达了。不应该所有启动的计时器都在任务关闭之前完成(至少这是我从 the code 得到的印象)？

能否请您帮助我理解这一点并使其更具确定性？

更新 #1，2018 年 9 月 23 日:

我还试验了事件时间窗口而不是处理时间窗口。如果我这样做:

val env = StreamExecutionEnvironment.getExecutionEnvironment
env
  .fromCollection(List("a", "b", "c", "d", "e"))
  .assignTimestampsAndWatermarks(new AscendingTimestampExtractor[String] {
    override def extractAscendingTimestamp(element: String): Long = {
      element.charAt(0).toInt
    }
  })

  .windowAll(TumblingEventTimeWindows.of(Time.seconds(1)))
  .trigger(EventTimeTrigger.create)
  .process(new ProcessAllWindowFunction[String, String, TimeWindow] {
    override def process(context: Context, elements: Iterable[String], out: Collector[String]): Unit = {
      out.collect(elements.toList.toString())
    }
  })

  .print()

env.execute()

然后又没有打印任何内容。调试器显示触发器的 onElement 被每个元素调用，但 onEventTime 从未被调用。

此外，如果我修改时间戳提取器以进行更大的步骤:

element.charAt(0).toInt * 1000

所有元素都被打印(每组一个元素，这是预期的)，除了最后一个。

更新 #2，2018 年 9 月 23 日:

更新 #1 在 this comment 中得到解答.

Answer 1

当有限源到达末尾时，如果您正在使用事件时间，则将注入(inject)带有时间戳 Long.MAX_VALUE 的 Watermark，这将导致所有事件时间计时器触发。但是，有了处理时间，Flink 会等待所有当前触发的定时器完成它们的 Action ，然后退出。

如您所料，您没有看到任何输出，因为源代码很快就完成了。

事件时间处理的确定性行为很简单；随着处理时间的增加，这实际上是无法实现的。

但这里有一个或多或少有效的技巧:

val env = StreamExecutionEnvironment.getExecutionEnvironment

val s = env.fromCollection(List("a", "b", "c", "d", "e"))
val t = env.addSource((context: SourceContext[String]) => {
  while(true) {
    Thread.sleep(100)
    context.collect("dummy")
  }
})

s.union(t)
  .filter(_ != "dummy")
  .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1)))
  .process(new ProcessAllWindowFunction[String, String, TimeWindow] {
    override def process(context: Context, elements: Iterable[String], out: Collector[String]): Unit = {
      out.collect(elements.toList.sorted.toString())
    }
  })
  .print()

env.execute()