[Feature Request]: Allowed write to custom sink from unbounded source

[lydian]

What would you like to happen?

I want to back up events from Kafka to S3 every minute. For example:

(
  pipeline 
  | "Read from Kafka" >> ReadFromKafka(consumer_config, topics, with_metadata=False)
  | "Transformed" >> beam.Map(my_transform)
  | "Add timestamp" >> beam.Map(add_timestamp)
  | "window" >> beam.WindowInto(window.FixedWindows(60))  # 1 mins
  | "writing to parquet" >> beam.io.WriteToParquet('s3://test-bucket/', pyarrow_schema)
)

However, when executing this pipeline, I got

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

Which is coming from https://github.com/apache/beam/blob/v2.41.0/sdks/python/apache_beam/io/iobase.py#L1145
where it always trying to create a global window. Also I found in here:

Currently only batch workflows support custom sinks.

The same function seems have no problem to be run in Java, but for python it is not supported. It would be nice to figured out a way to support this.

Issue Priority

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

Issue Components

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

[shunping]

+1

[liferoad]

The work related to #34777 should solve this issue

[liferoad]

cc @razvanculea