PubsubIO在使用DataflowRunner和Dataflow服务运行时，不将自定义时间戳属性输出为context.timestamp

Question

我正在研究一个Apache Beam项目，该项目遇到了与自定义时间戳属性相关的Dataflow服务和PubsubIO问题。Beam SDK的当前版本是2.7.0。

在该项目中，我们有两个通过PubSub主题和订阅进行通信的数据流作业：

第一条管道（将数据下沉到PubSub）

这个管道工作在每个基础的消息上，因此除了globalwindows（Beam默认）之外，它没有应用自定义窗口策略。在这个管道的末尾，我们使用pubsubio.writeMessages()将已经分配了包括事件时间戳（例如“published_at”）的属性映射的所有消息下沉（写入）到一个PubSubTopic。

注意：如果我们使用pubsubio.writeMessages().WithTimestampatAttribute()，此方法将告诉pubsubio.shardfn、pubsubio.writefn和pubsubclient将下沉管道的处理时间写入/覆盖到映射中的该属性。

第二个管道（从PubSub读取数据）

在第二个管道（读取管道）中，我们尝试了pubsubio.readmessageswithattributes().withtimestampattribute（“published_at”）和pubsubio.readstrings().withtimestampattribute（“published_at”）。

• 使用DirectRunner运行时，一切都如预期的那样工作良好。这些消息从PubSub订阅中读取并输出到下游阶段，其processContext.timestamp()等于其事件时间戳“published_at”.
• 但是当使用DataflowRunner运行时，processContext.timestamp()总是被设置为接近实时，接近下沉管道的处理时间。我们检查并可以确认这些时间戳不是来自PubSub的发布时间。然后，所有数据都被分配到错误的窗口，与它们的事件域时间戳相比较。我们希望删除较晚的数据，而不是将其分配到无效的窗口中。

注意：在打开第二个管道以获得某种历史/后期数据之前，我们已经让Pubsub主题填充了大量数据。

具有无效上下文时间戳的Pubsub消息

假定根本原因

深入研究DataflowRunner的源代码，我们可以看到Dataflow服务使用一个完全不同的Pubsub代码（在构建管道时重写pubsubio.Read）从Pubsub读取和接收到Pubsub。

因此，如果我们想要使用Beam SDK的PubsubIO，我们必须使用实验选项“enable_custom_pubsub_source”。但是到目前为止还没有运气，因为我们遇到了这个问题https://jira.apache.org/jira/browse/beam-5674，并且还没有能够测试Beam SDK的Pubsub代码。

变通解决方案

我们当前的解决方案是，在为消息分配窗口的步骤之后，我们实现了dofn来根据它们的intervalwindow检查它们的事件时间戳。如果窗口无效，那么我们只需删除消息，然后每周或半周运行一次作业，从历史来源更正它们。最好有一些缺失的数据，而不是计算不当的数据。

由于无效的窗口而丢弃的消息

请与我们分享这个案例的经验。我们知道，从数据流水印管理的角度来看，如果摄入的数据稀疏（不够密集），水印就会自动调整到当前的实时状态。

我们还认为，我们对Dataflow服务维护PubSubUnboundedSource的输出时间戳的方式有些误解，因为我们对Apache Beam和Google的Dataflow仍然是新手，所以有些事情我们还不知道。

多谢！

Answer 1

我找到了解决这个问题的方法。在我的下沉管道中，timestamp属性设置的日期格式与RFC3339标准相比是错误的。格式化日期缺少“Z”字符。我们要么修复了'z'字符，要么改为使用自纪元以来的毫秒。两者都很有效。

但有一点是，当Dataflow服务无法解析错误的日期格式时，它确实警告或抛出错误，但却占用了所有元素的处理时间，因此，它们被分配到错误的event_time窗口。