如何在 Apache Beam Python 中获取 window 时间戳的结尾

Question

我正在从批处理的日志数据中创建每 5 秒 20 秒的滑动时间 windows：

    rows = p | 'read events' >> beam.io.Read(beam.io.BigQuerySource(query=query))

    # set timestamp field used for windowing and set 20 second long window every 5 seconds
    ts_rows = (rows | 'set timestamp' >> beam.ParDo(AddTimestampDoFn())
                    | 'set window' >> beam.WindowInto(window.SlidingWindows(20,5)))

    # extract only user id and relevant data, group and process
    rows_with_data = (ts_rows | 'extract data' >> beam.FlatMap(lambda row: 
                                [(str(row['user_id']),[row['data1'], row['data2'],row['data3']])])
                              | 'group by user id' >> beam.GroupByKey()
                              | 'Process window' >> beam.ParDo(WindowDataProcessingDoFn()))

如何访问 Python 中每个 window 的时间戳信息？（Java 的答案在这里，但我不知道如何将其翻译成 Python：How to get the max timestamp of the current sliding window）理想情况下，我希望每个 window 的结束时间比 window 内数据的最大或最小时间戳。

Answer 1

我去了你提供的link

注意：window=beam.DoFn.WindowParam 是您 link 页面上提到的参数。

window结束时间是beam.DoFn.WindowParam.end。在Python中，您可以这样访问它：

定义您的 DoFn：

class BuildRecordFn(beam.DoFn):
def __init__(self):
    super(BuildAdsRecordFn, self).__init__()

def process(self, element,  window=beam.DoFn.WindowParam):
    #window_start = window.start.to_utc_datetime()
    window_end = window.end.to_utc_datetime()
    return [element + (window_end,)]

然后像这样使用它：

    lines = p | ReadFromText(known_args.input)
    counts = (
        lines
        | 'ParseEventFn' >> beam.ParDo(ParseEventFn())

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

        | 'ExtractObjectID' >> beam.Map(lambda elem: (elem["objectID"]))

        | 'FixedWindow' >> beam.WindowInto(
            beam.window.FixedWindows(60*1))

        | 'PairWithOne' >> beam.Map(lambda x: (x, 1))

        | 'GroupAndSum' >> beam.CombinePerKey(sum)

        | 'AddWindowEndTimestamp'(beam.ParDo(BuildRecordFn()))

        | 'Format' >> beam.Map(format_result)

        | WriteToText(known_args.output) 


    def format_result(xs):
        ys = [str(x) for x in xs]
        return ','.join(ys)

如何在 Apache Beam Python 中获取 window 时间戳的结尾

How to get the end of window timestamp in Apache Beam Python

python

google-cloud-dataflow

apache-beam