如何在 Apache Beam 中提取 Google PubSub 发布时间

Question

我的目标是能够访问由 Google PubSub 在 Apache Beam（数据流）中记录和设置的 PubSub 消息发布时间。

    PCollection<PubsubMessage> pubsubMsg
            = pipeline.apply("Read Messages From PubSub",
            PubsubIO.readMessagesWithAttributes()
                .fromSubscription(pocOptions.getInputSubscription()));

似乎没有包含一个属性。我试过了

 .withTimestampAttribute("publish_time")

也不走运。我错过了什么？是否可以在数据流中提取 Google PubSub 发布时间？

Answer 1

Java版本:

PubsubIO 将从Pub/Sub读取消息并将消息发布时间分配给元素作为记录时间戳。因此，您可以使用 ProcessContext.timestamp() 访问它。例如：

p
    .apply("Read Messages", PubsubIO.readStrings().fromSubscription(subscription))
    .apply("Log Publish Time", ParDo.of(new DoFn<String, Void>() {
        @ProcessElement
        public void processElement(ProcessContext c) throws Exception {
            LOG.info("Message: " + c.element());
            LOG.info("Publish time: " + c.timestamp().toString());
            Date date= new Date();
            Long time = date.getTime();
            LOG.info("Processing time: " + new Instant(time).toString());
        }
    }));

我提前发布了一条消息（事件和处理时间之间存在显着差异）并且 DirectRunner 的输出是：

Mar 27, 2019 11:03:08 AM com.dataflow.samples.LogPublishTime processElement
INFO: Message: I published this message a little bit before
Mar 27, 2019 11:03:08 AM com.dataflow.samples.LogPublishTime processElement
INFO: Publish time: 2019-03-27T09:57:07.005Z
Mar 27, 2019 11:03:08 AM com.dataflow.samples.LogPublishTime processElement
INFO: Processing time: 2019-03-27T10:03:08.229Z

最小代码here

Python版本:

现在可以通过 process 方法的 DoFn.TimestampParam 访问时间戳 (docs):

class GetTimestampFn(beam.DoFn):
  """Prints element timestamp"""
  def process(self, element, timestamp=beam.DoFn.TimestampParam):
    timestamp_utc = datetime.datetime.utcfromtimestamp(float(timestamp))
    logging.info(">>> Element timestamp: %s", timestamp_utc.strftime("%Y-%m-%d %H:%M:%S"))
    yield element

注意：日期解析感谢this answer。

输出：

INFO:root:>>> Element timestamp: 2019-08-12 20:16:53

满code

如何在 Apache Beam 中提取 Google PubSub 发布时间

How to extract Google PubSub publish time in Apache Beam

google-cloud-platform

google-cloud-dataflow

apache-beam