google-cloud-dataflow - Apache 光束 : why is the timestamp of aggregate value in Global Window 9223371950454775?

Question

我们从 Google Dataflow 1.9 迁移到 Apache Beam 0.6。我们注意到应用全局窗口后时间戳的行为发生了变化。在 Google Dataflow 1.9 中，我们将在窗口/组合函数之后在 DoFn 中获得正确的时间戳。现在我们得到了时间戳的一些巨大值，例如9223371950454775，Apache Beam 版本中全局窗口的默认行为是否发生变化？

input.apply(name(id, "Assign To Shard"), ParDo.of(new AssignToTest()))
      .apply(name(id, "Window"), Window
          .<KV<Long, ObjectNode >>into(new GlobalWindows())
          .triggering(Repeatedly.forever(
              AfterProcessingTime
                  .pastFirstElementInPane()
                  .plusDelayOf(Duration.standardMinutes(1))))
          .discardingFiredPanes())
      .apply(name(id, "Group By Shard"), GroupByKey.create())
      .appy(.....) }

Answer 1

TL;DR:当您组合一堆带时间戳的值时，您需要为聚合结果选择一个时间戳。对于这个输出时间戳有多个很好的答案。在 Dataflow 1.x 中，默认值是输入时间戳的最小值。根据我们在 Beam 1.x 中的经验，默认值已更改为窗口末尾。您可以通过将 .withTimestampCombiner(TimestampCombiner.EARLIEST) 添加到 Window 转换来恢复之前的行为。

<小时/>

我会打开这个。让我们使用 @ 符号将一个值及其时间戳配对。仅关注一个键，您就有带时间戳的值 v1@t1、v2@t2，...等。我将继续使用您的原始 示例GroupByKey 即使这也适用于组合值的其他方式。因此值的输出可迭代是 [v1, v2, ...] 以任意顺序。

以下是时间戳的一些可能性:

• 分钟(t1，t2，...)
• max(t1, t2, ...)
• 这些元素所在窗口的末尾(忽略输入时间戳)

这些都是正确的。这些都可以作为 Dataflow 1.x 中的 OutputTimeFn 和 Apache Beam 中的 TimestampCombiner 的选项。

时间戳有不同的解释，它们对于不同的事情很有用。聚合值的输出时间控制下游水印。因此，选择较早的时间戳可以更好地保留下游水印，而较晚的时间戳则可以使其继续前进。

• min(t1, t2, ...) 允许您解压可迭代并重新输出v1@t1
• max(t1, t2, ...) 准确地模拟聚合值完全可用的逻辑时间。由于实现细节的原因，Max 确实往往是最昂贵的。
• 窗口结束:
  • 模拟了该聚合代表窗口所有数据的事实
  • 非常容易理解
  • 允许下游水印尽快前进
  • 非常高效

出于所有这些原因，我们将默认值从分钟切换为窗口结束。

在 Beam 中，您可以通过将 .withTimestampCombiner(TimestampCombiner.EARLIEST) 添加到 Window 转换来恢复之前的行为。在 Dataflow 1.x 中，您可以通过添加 .withOutputTimeFn(OutputTimeFns.outputAtEndOfWindow()) 迁移到 Beam 的默认设置。

另一个技术细节是用户定义的 OutputTimeFn 被删除并被 TimestampCombiner 枚举取代，因此只有这三个选择，而不是一个完整的 API 来编写您的拥有。