什么实际管理光束中的水印？

Question

Beam 的强大功能来自于它的高级窗口功能，但它也有点令人困惑。

在本地测试中看到一些奇怪的地方（我使用 rabbitmq 作为输入源），其中消息并不总是得到 ackd，并且修复了 windows 并不总是关闭的问题，我开始挖掘围绕 Whosebug 和 Beam 代码库。

似乎存在关于何时设置水印的特定于源的问题：

• RabbitMQ水印不推进：Apache Beam : RabbitMqIO watermark doesn't advance
• PubSub 水印不会针对低容量推进：https://issues.apache.org/jira/browse/BEAM-7322
• SQS IO 在没有新传入消息的一段时间内不会推进水印 - https://github.com/apache/beam/blob/c2f0d282337f3ae0196a7717712396a5a41fdde1/sdks/java/io/amazon-web-services/src/main/java/org/apache/beam/sdk/io/aws/sqs/SqsIO.java#L44

（和其他人）。此外，似乎有 Checkpoints (CheckpointMarks) 的独立概念与 Watermarks.

相反

所以我想这是一个多部分问题：

1. 什么代码负责移动水印？它似乎是 Source 和 Runner 的某种组合，但我似乎无法 find 它以更好地理解它（或针对我们的用例调整它）。这对我来说是一个特殊的问题，因为在低音量期间水印永远不会前进并且消息不是 ackd
2. 我没有看到太多关于 Checkpoint/Checkpoint 标记在概念上是什么的文档（非代码 Beam 文档没有讨论它）。 CheckpointMark 如何与 Watermark 交互（如果有）？

Answer 1

1. 每个 PCollection 都有自己的水印。水印表明 特定 PCollection 的完整性。源负责它生成的 PCollection 的水印。水印到下游 PCollections 的传播是自动的，无需额外的近似；可以大致理解为"the minimum of input PCollections and buffered state"。因此，在您的情况下，需要 RabbitMqIO 查看水印问题。我不熟悉这个特定的 IO 连接器，但是如果您还没有这样做，那么向用户列表发送错误报告或电子邮件会很好。
2. 检查点是一段特定于源的数据，只要运行者持久保留检查点，它就可以继续读取而不会丢失消息。消息 ACK 往往发生在检查点完成时，因为运行器在知道消息永远不需要重新读取时调用此方法。