在批处理管道中，如何为批处理源中的数据分配时间戳，例如 Beam 管道中的 csv 文件答案

【问题标题】：In a batch pipeline how do I assign timestamps to data from the batch sources for example csv files in a Beam pipeline在批处理管道中，如何为批处理源中的数据分配时间戳，例如 Beam 管道中的 csv 文件
【发布时间】：2019-06-25 15:11:39
【问题描述】：

我正在批处理管道中从有界源（csv 文件）读取数据，并希望根据存储为 csv 文件中的列的数据为元素分配时间戳。如何在 Apache Beam 管道中执行此操作？

【问题讨论】：

标签： google-cloud-dataflow apache-beam

【解决方案1】：

如果您的批处理数据源包含每个元素的基于事件的时间戳，例如，您有一个包含元组 {'timestamp, 'userid','ClickedSomething'} 的点击事件。您可以将时间戳分配给管道中 DoFn 内的元素。

Java：

public void process(ProcessContext c){
     c.outputWithTimestamp(
         c.element(), 
         new Instant(c.element().getTimestamp()));
}

Python：

'AddEventTimestamps' >> beam.Map(
            lambda elem: beam.window.TimestampedValue(elem, elem['timestamp']))

[从 Beam 指南编辑非 lambda Python 示例：]

class AddTimestampDoFn(beam.DoFn):

  def process(self, element):
    # Extract the numeric Unix seconds-since-epoch timestamp to be
    # associated with the current log entry.
    unix_timestamp = extract_timestamp_from_log_entry(element)
    # Wrap and emit the current entry and new timestamp in a
    # TimestampedValue.
    yield beam.window.TimestampedValue(element, unix_timestamp)

timestamped_items = items | 'timestamp' >> beam.ParDo(AddTimestampDoFn())

[根据安东评论编辑] 更多信息可以找到@

https://beam.apache.org/documentation/programming-guide/#adding-timestamps-to-a-pcollections-elements

【讨论】：

一些相关信息也可以在指南中找到：beam.apache.org/documentation/programming-guide/…