Write just one file per window with Apache Beam Python `WriteToFiles` transform

user_12917806 0
Viewed 1167

need some help. I have some trivial task reading from Pub/Sub and write to batch file in GCS, but have some struggle with 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))

I need one file after my window fires, but the number of files is equal to the number of messages from Pubsub, can anyone help me? current output files but i need only one file.

apache-beam google-cloud-dataflow python
2 Answers

I recently ran into this issue and dug into the source code:

fileio.WriteToFiles will attempt to output each element in the bundle as an individual file. WireToFiles will only fallback to writing to sharded files if the number of elements exceeds the number of writers.

To force a single file to be created for all elements in the window, set max_writers_per_bundle to 0

WriteToFiles(shards=1, max_writers_per_bundle=0)
JMM_12990

You can pass WriteToFiles(shards=1, ...) to limit the output to a single shard per window and triggering.

user_9493270
Related