关于谷歌云数据流：如何使用 Apache Beam 在 Python 中将有界 pcollection 转换为无界？

    known_args, pipeline_options = parse_args(sys.argv)
    pipeline = beam.Pipeline(options=pipeline_options)
    lines = pipeline | ‘read’ >> beam.io.ReadFromText(known_args.input)
    parsed_json = lines //
      | ‘cut’ >> beam.Map(lambda x: x[39:])//
      | ‘jsonify’ >> beam.Map(json_loads).with_outputs(‘invalid_input’, main=‘main’)

    keyed_lines = parsed_json[‘main’]//
      | ‘key_filter’ >> beam.Filter(lambda x: u‘key’ in x)//
      | ‘keyer’ >> beam.Map(lambda x: (x[‘key’], x))//
      | ‘log_processed_rows’ >> beam.Map(log_processed_rows)

    window_trigger = trigger.DefaultTrigger()

    windowed_lines = keyed_lines//
       | ‘timestamp’ >> beam.ParDo(AddTimestampDoFn())//
       | ‘window’ >> beam.WindowInto(beam.window.SlidingWindows(size=3600+600, period=3600), trigger=window_trigger,
                                     accumulation_mode=trigger.AccumulationMode.DISCARDING) # 1 hour steps with 10 minutes of overlap

    results = windowed_lines //
        | ‘groupbykey’ >> beam.GroupByKey()//
        | ‘parse’ >> beam.ParDo(ParseSendingsDoFn()).with_outputs(‘too_few_rows’, ‘invalid_rows’, ‘missing_recipients’, main=‘main_valid’)  

    output = results[‘main_valid’] //
        | ‘format’ >> beam.Map(output_format)//
        | ‘write’ >> beam.io.WriteToText(known_args.output, file_name_suffix=".gz")