unbounded PCollections writes to files support in iobase derived IOs: AvroIO, ParquetIO , TextIO, TFRecordIO by razvanculea · Pull Request #34777 · apache/beam

razvanculea · 2025-04-29T13:11:34Z

iobase patch for writing unbounded PColl sources.
Implement unbounded PCollections writes to files support in iobase derived IOs: AvroIO, ParquetIO , TextIO, TFRecordIO
Tests for streming PCollections writes to files in AvroIO, ParquetIO , TextIO, TFRecordIO

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

…IO, ParquetIO and TFRecordIO

…ming writes for AvroIO, ParquetIO and TFRecordIO

sdks/python/apache_beam/io/iobase.py

github-actions · 2025-04-29T14:39:20Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

sjvanrossum · 2025-04-29T14:56:30Z

sdks/python/apache_beam/io/iobase.py

+        init_result_window_coll = (
          keyed_pcoll
-          | core.WindowInto(window.GlobalWindows())
-          | core.GroupByKey()
-          | 'WriteBundles' >> core.ParDo(
-              _WriteKeyedBundleDoFn(self.sink), AsSingleton(init_result_coll)))
+            | 'Pair init' >> core.Map(lambda x: (None, x))
+            | 'Pair init gbk' >> core.GroupByKey()
+            | 'InitializeWindowedWrite' >> core.Map(
+              lambda _, sink: sink.initialize_write(), self.sink)


This doesn't have to depend on the main input, right?

sjvanrossum · 2025-04-29T15:02:17Z

sdks/python/apache_beam/io/iobase.py

+        if isinstance(pcoll.windowing.windowfn, window.GlobalWindows):
+          widowed_pcoll = (
+            pcoll
+              | core.WindowInto(window.FixedWindows(self.sink.triggering_frequency),
+                                trigger=beam.transforms.trigger.AfterWatermark(),
+                                accumulation_mode=beam.transforms.trigger.AccumulationMode.DISCARDING,
+                                allowed_lateness=beam.utils.timestamp.Duration(seconds=0))
+          )


If the input is globally windowed a trigger within the global window set to repeat every sink.triggering_frequency would be what I'd expect instead of windowing into FixedWindows.

that would be an alternative way to implement.
The windowing will help have elements of the same window in the same file, vs group into batches & trigger will have elements as they come (possibly mixed)

github-actions · 2025-04-29T15:07:23Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @shunping for label python.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

fix write_ptransform_test.py fix iobase.Sink.finalize_write params fix lint warnings reorder imports

liferoad · 2025-05-02T19:36:56Z

is this PR tested with DataflowRunner?

razvanculea · 2025-05-05T08:17:39Z

is this PR tested with DataflowRunner?

@liferoad i've added the iobase_it_test from PR 34814 and it runs on Dataflow (it's actually using TextIO for write), but i don't have an IT for TextIO/parquetIO/AvroIO/TFRecordIO, I can try to add them, but i can't use TestStream in Dataflow, so i'll need to use PubSub public topics , right ?

liferoad · 2025-05-05T14:44:55Z

is this PR tested with DataflowRunner?

@liferoad i've added the iobase_it_test from PR 34814 and it runs on Dataflow (it's actually using TextIO for write), but i don't have an IT for TextIO/parquetIO/AvroIO/TFRecordIO, I can try to add them, but i can't use TestStream in Dataflow, so i'll need to use PubSub public topics , right ?

I forgot whether periodic impulse can trigger #25598. You could try that. And better add some checks about the output results.

razvanculea · 2025-05-05T18:59:59Z

is this PR tested with DataflowRunner?

@liferoad i've added the iobase_it_test from PR 34814 and it runs on Dataflow (it's actually using TextIO for write), but i don't have an IT for TextIO/parquetIO/AvroIO/TFRecordIO, I can try to add them, but i can't use TestStream in Dataflow, so i'll need to use PubSub public topics , right ?

is this PR tested with DataflowRunner?

@liferoad i've added the iobase_it_test from PR 34814 and it runs on Dataflow (it's actually using TextIO for write), but i don't have an IT for TextIO/parquetIO/AvroIO/TFRecordIO, I can try to add them, but i can't use TestStream in Dataflow, so i'll need to use PubSub public topics , right ?

I forgot whether periodic impulse can trigger #25598. You could try that. And better add some checks about the output results.

@liferoad, I added an IT in parquetio_it_test.py using PeriodicImpulse and it's working fine on Dataflow

Did not add IT for AvroIO/TFRecordIO because there is no IT at all. I could do like for parquetIO later, if we deem this really a plus.

…requency param. On GlobalWindow throw an exception if trigger_frequency is not set / 0.

razvanculea · 2025-05-07T11:48:23Z

Improved support for user windowing.
If an unbounded PCollection with GlobalWindow is written:

without setting triggering_frequency => throw an exception
if triggering_frequency is set, then the output will be written per requested frequency

If an unbounded PCollection with a user set windowing is written:

without setting triggering_frequency => then the output will be written per user defined window
if triggering_frequency is set, then the output will be written per requested frequency (this overrides the user defined window)

If num_shards is set : per triggering_frequency/user window, the output is written in num_shards files
If num_shards is not set : per triggering_frequency/user window, the output is written with 1 file per bundle

github-actions · 2025-05-15T12:15:20Z

Reminder, please take a look at this pr: @shunping

github-actions · 2025-05-19T12:15:56Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @jrmccluskey for label python.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

shunping · 2025-05-20T18:19:30Z

@razvanculea, sorry about the delay of the review. Do you mind merging to the latest master so that the tests are re-triggered?

shunping · 2025-05-20T18:20:08Z

waiting on author

razvanculea · 2025-05-22T11:52:34Z

@shunping don't review this anymore.
I'll break it into IObase+FileBasedSync PR, and 1 PR per IO (Parquet,Avro,TFRecord,Text)
As @sjvanrossum pointed out I could avoid breaking changes, in case there are any private developped IOs that currently work for batch. These would not work for streaming without changes, but would still work as before on batch with newer SDK.

shunping · 2025-05-25T11:51:21Z

sg! Feel free to tag me when they are ready.

github-actions · 2025-06-01T12:14:24Z

Reminder, please take a look at this pr: @jrmccluskey

razvanculea added 2 commits April 29, 2025 12:15

iobase patch for unbounded PColl and adding streaming writes for Avro…

0e0cb0b

…IO, ParquetIO and TFRecordIO

update changes.md : iobase patch for unbounded PColl and adding strea…

eb3bb71

…ming writes for AvroIO, ParquetIO and TFRecordIO

github-actions bot added python examples io labels Apr 29, 2025

Merge branch 'master' into iobase

3c720d3