使用 Apache Beam Python `WriteToFiles` 转换每个窗口只写一个文件

Question

需要一些帮助。我有一些从 Pub/Sub 读取并写入 GCS 中的批处理文件的琐碎任务，但是在使用 fileio.WriteToFiles 时遇到了一些困难

with beam.Pipeline(options=pipeline_options) as p:
  input = (p | 'ReadData' >> beam.io.ReadFromPubSub(topic=known_args.input_topic).with_output_types(bytes)
             | "Decode" >> beam.Map(lambda x: x.decode('utf-8'))
             | 'Parse' >> beam.Map(parse_json)
             | ' data w' >> beam.WindowInto(
                 FixedWindows(60),
                 accumulation_mode=AccumulationMode.DISCARDING
             ))

  event_data = (input
             | 'filter events' >> beam.Filter(lambda x: x['t'] == 'event')
             | 'encode et' >> beam.Map(lambda x: json.dumps(x))
             | 'write events to file' >> fileio.WriteToFiles(
                    path='gs://extention/ga_analytics/events/', shards=0))

窗口触发后我需要一个文件，但文件数等于来自 Pubsub 的消息数，有人可以帮助我吗？ 当前输出文件， 但我只需要一个文件。

Answer 1

我最近遇到了这个问题并挖掘了源代码：

fileio.WriteToFiles 将尝试将包中的每个元素作为单独的文件输出。如果元素数量超过写入者数量，WireToFiles 只会回退到写入分片文件。

要强制为窗口中的所有元素创建单个文件，请将 max_writers_per_bundle 设置为 0

WriteToFiles(shards=1, max_writers_per_bundle=0)