Flink窗口和后期事件

Question

我有一个用例，需要以不同的方式处理延迟事件和正常事件：如果事件在其窗口关闭后到达，则应将其发送到另一个路径。

我想是这样的。sideOutputLateData（..） 可以帮我解决这个问题。在正常情况下（即使用真实世界的数据）也是如此。但如果我想用伪造的数据来测试它，它就会停止工作。

我希望类似于：

val env = StreamExecutionEnvironment.createLocalEnvironment()
env.setParallelism(1)

val events: DataStream[(Int, Long)] = env.fromElements(
  (1, 1),
  (1, 15),
  (1, 25),
  (1, 8) //late Event
)
val lateEvents = OutputTag[(Int, Long)]("lateEvents")

val windowedSum = events
  .assignAscendingTimestamps(e => e._2)
  .windowAll(TumblingEventTimeWindows.of(time.Time.milliseconds(10)))
  .sideOutputLateData(lateEvents)
  .sum(position=0)

val lateEventsStream = windowedSum
  .getSideOutput(lateEvents)
  // Handle differently
  .map(e => (e._1 + 100, e._2))

windowedSum.print()
lateEventsStream.print()
// execute program
env.execute("Flink Scala watermarking test")

将导致：

[info] (1,1)
[info] (1,15)
[info] (1,25)
[info] (101, 8)

相反，我得到了：

[info] (2,1)
[info] (1,15)
[info] (1,25)

如果我使用socketTextStream作为具有相同数据的源，它将按预期工作。

这告诉我，在数据输入速度非常快的情况下，水印并没有像应该的那样前进。我尝试将setAutoWatermarkInterval调整为一个非常小的值，但没有成功。

我错过了什么吗？我如何测试我的工作？

Answer 1

感谢@Dominik Wosinski为我指明了正确的方向。对于在Flink文档的复杂性中迷失的其他人，我将在这里发布我的解决方案：

正如怀疑的那样，问题是快速输入数据没有推进水印。这是因为默认情况下，Flink将每200ms检查一次水印是否应该推进。您可以使用

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.getConfig.setAutoWatermarkInterval(10) // or even lower

显然，这对于作为4元素列表的快速输入是不够的。

解决方案是在每次事件时发出水印（请注意，在生产环境中不建议这样做）。

要实现这种解决方案，我们需要扩展WatermarkGenerator类：

class MyPunctuatedWatermarkAssigner extends WatermarkGenerator[(Int, Long)] {

  override def onEvent(
      event: (Int, Long),
      eventTimestamp: Long,
      output: WatermarkOutput
  ): Unit = {
    // emit at every event
    output.emitWatermark(new Watermark(event._2))
  }

  // do nothing at AutoWatermarkInterval
  override def onPeriodicEmit(output: WatermarkOutput): Unit = {}
}

要将此生成器分配给流，我们首先需要创建水印策略：

class MyStrategy extends WatermarkStrategy[(Int, Long)] {
  override def createWatermarkGenerator(
      context: WatermarkGeneratorSupplier.Context
  ): WatermarkGenerator[(Int, Long)] = new MyPunctuatedWatermarkAssigner
}

（此类还可以实现可选的CreateTimestAssigner方法）

然后我们可以在流中使用它：

eventsStream
   .assignTimestampsAndWatermarks(new MyStrategy())

Answer 2

您可以尝试使用水印策略，该策略将为每个元素发出水印，而不是定期生成它们。实现水印生成器时，可以通过在元素内部发出水印来实现这一点，如下所述。这是用于测试的最佳和最可靠的方法。