[Bug]: FileBasedSink seems to not work on streaming

[damccorm]

What happened?

I tried running a few different pipelines using ios based on filebasedsink. They all failed on Dataflow and/or locally with various errors when the --streaming flag is used. Adding a few here:

Writing to parquet:

    with beam.Pipeline(options=beam_options) as pipeline:
        elements = pipeline | "Create elements" >> beam.Create(["Hello", "World!"])
        import pyarrow
        elements | output | beam.io.WriteToParquet(
        "gs://path",
        schema=pyarrow.schema(
            [("name", pyarrow.binary()), ("age", pyarrow.int64())]
        ),
    )

Writing to textio:

    with beam.Pipeline(options=beam_options) as pipeline:
        elements = pipeline | "Create elements" >> beam.Create(["Hello", "World!"])
        elements | beam.io.WriteToText("./test")

This works locally, but fails on Dataflow before launch (unclear if it is the fault of transform or some translation layer, seems like it comes from this particular AsSingleton call -

beam/sdks/python/apache_beam/io/iobase.py

Line 1170 in 18849de

AsSingleton(pre_finalize_coll)).with_output_types(str)

Writing to textio with periodic impulse source gives a different error:

    with beam.Pipeline(options=beam_options) as pipeline:
        elements = pipeline | "Create elements" >> PeriodicImpulse()  | 'ApplyWindowing' >> beam.WindowInto(
          beam.transforms.window.FixedWindows(20))
        elements | beam.io.WriteToText("./test")

ValueError: GroupByKey cannot be applied to an unbounded PCollection with global windowing and a default trigger

This isn't correct since this is clearly windowed before the write.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[liferoad]

Please use WriteToFiles.

[razvanculea]

@damccorm pre my IObase work, all IOs derived from IObased and FileBasedSink don't work in streaming (i should have logged an issue). They do work in batch.

My PR to add streaming support:
PR - [(#35137)] adds support in IObase (fyi, most of the implementation needs to be in FileBasedSink for most used IOs : Text,Avro,Parquet,TFRecords)
PR - [(#35253)] adds surpport in FileBasedSink and TextIO
PR - [(#35418)] , i added support for all the others IOs using IObase and FileBasedSink in apache_beam : AvroIO, ParquetIO, TFRecordsIO

Your fist 2 tests are ok for batch, they would fail in streaming (with my PRs as well) because window is Global and no triggering frequency specified in the Write.

Your last test looks correct and works with my PRs, code (in the attached achive as test_periodimpulse.py)
See job id in my internal test project 2025-08-07_05_14_41-9611060739629941282

#Danny McCormick test
  elements =(
      p 
      | "Create elements" >> beam.transforms.periodicsequence.PeriodicImpulse(
          start_timestamp=1, 
          stop_timestamp=100, 
          fire_interval=10, 
          apply_windowing=False )
      | 'ApplyWindowing' >> beam.WindowInto(FixedWindows(20))
      | beam.io.WriteToText("./test")
  )

I think i'll add thses examples under sdks/python/apache_beam/examples/unbouded_sinks/

unbounded_sinks.zip

[github-actions]

This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.