google-cloud-dataflow - 在 GCD 上运行的 PubsubIO 的水印启发式是什么？

Question

嗨，我正在尝试运行一个管道，我正在计算以 30 秒心跳*(10K 流，每个心跳每 30 秒)发布到 pubsub 的消息之间的差异。我不关心 100% 的数据完整性，但我想了解 PubsubIO 的水印启发式是什么(如果我可以调整它)，以确定我是否可以忽略损失足够低的后期数据。

*请注意，pubsub 主题提供了 [可能几天的] 持久性，以防我们必须关闭管道，因此启发式方法在积压订阅中运行良好非常重要。

有人可以解释一下水印是如何计算的(假设使用了 timestamplabel() )，以及如何调整它(如果有的话)？

Answer 1

以下是我们如何计算 PubSub 水印的简要说明:

我们的目标是为通过 PubSub 发送到我们的流媒体管道的数据构建合理的启发式水印。我们对将数据发送到 PubSub 的源做了一些假设。具体来说，我们假设原始数据的时间戳“表现良好”，换句话说，在将源数据发送到 PubSub 之前，我们期望源数据上有一定数量的我们的订单时间戳。任何以超出允许的乱序范围的时间戳发送的数据都将被视为延迟数据。在我们当前的实现中，这个界限是 10 秒 这意味着在发送到 pubsub 之前重新排序时间戳最多 10 秒不会创建延迟数据。我们称这个值为估计带。然后，构建 PubSub 水印的问题简化为确保没有额外的数据因通过 PubSub 传输而延迟。

我们在 PubSub 中面临哪些挑战？由于 pubsub 不保证排序，我们必须有某种额外的元数据来充分了解积压。幸运的是，PubSub 提供了“最旧的未确认发布时间戳”方面的积压测量。这与我们消息的事件时间戳不同，因为 PubSub 与通过它发送的应用程序级元数据无关，而是 PubSub 摄取消息时的时间戳。

虽然这种测量听起来类似于水印，但它并不相同。我们不能简单地使用最旧的未确认发布时间戳作为水印。这些时间戳不等于事件时间戳，在发送历史(过去)数据的情况下，它可能是任意远的。这些时间戳的排序也可能不同，因为如上所述，我们允许进行有限的重新排序。但是，我们可以将其用作积压的度量，以了解有关积压中存在的事件时间戳的足够信息，以便我们可以建立合理的水印，如下所示。

我们将数据到达的订阅称为基本订阅。看看我们的基本订阅，我们看到消息可能无序到达。我们用它的 pubsub 发布时间戳“pt”和它的事件时间时间戳“et”来标记每条消息。请注意，两个时域可以不相关



基本订阅上的一些消息未被确认形成积压。这可能是由于它们尚未交付，或者它们可能已交付但尚未处理。还请记住，从此订阅中提取的内容分布在多个分片中。因此，不可能仅通过查看基本订阅就知道我们的水印应该是什么。

我们继续创建第二个仅元数据跟踪订阅，用于有效检查基本订阅的积压，并在积压中获取事件时间戳的最小值。通过在跟踪订阅上保持很少或没有积压，我们可以在基本订阅最旧的未确认消息之前检查消息。



我们通过确保从该订阅中提取在计算上是廉价的，从而跟上跟踪订阅。相反，如果我们在跟踪订阅方面落后太多，我们将停止推进水印。为此，我们确保至少满足以下条件之一:

跟踪订阅充分超前于基本订阅，充分超前意味着跟踪订阅至少超前估计频带。这确保了估计带内的任何有界重新排序都被考虑在内。

跟踪订阅足够接近实时。换句话说，跟踪订阅没有积压。

一旦我们持久保存了有关消息发布和事件时间戳的元数据，我们就会尽快确认跟踪订阅上的消息。我们以稀疏直方图格式存储此元数据，以最大限度地减少使用的空间量和持久写入的大小。

最后，我们确保我们有足够的数据来进行合理的水印估计。我们采用一系列事件时间戳和范围内的发布时间戳

 [ min ( base sub oldest unack'd, tracking sub oldest unack'd - 10 sec) , 
         tracking sub oldest unack'd ]

这确保我们考虑积压中的所有事件时间戳，或者如果积压很小，则最近的估计带，以进行水印估计。

最后，水印值被计算为带中的最小事件时间。

另请注意，此方法是正确的，但会产生过于保守的水印。由于我们考虑了跟踪订阅中基本订阅最旧的未确认消息之前的所有消息，因此我们可能会在已确认消息的水印估计中包含事件时间戳。

此外，还有一些启发式方法可以确保取得进展。上述方法在密集、频繁到达的数据的情况下效果很好。在数据稀疏或不频繁的情况下，可能没有足够的近期消息来建立合理的估计。如果我们没有看到超过 的订阅数据两分钟 (并且没有积压)，我们将水印推进到接近实时。这确保即使没有更多消息即将到来，水印和管道也会继续取得进展。

以上都保证了只要源数据事件时间戳重新排序在估计范围内，就不会有额外的迟到数据。