{"record":{"id":"03855842f9d555b3","repo":"apache/beam","slug":"to-write-a-globalwindow-unbounded-pcollection-triggering","errorCode":null,"errorMessage":"To write a GlobalWindow unbounded PCollection, triggering_frequency must be set and be greater than 0","messagePattern":"To write a GlobalWindow unbounded PCollection, triggering_frequency must be set and be greater than 0","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/iobase.py","lineNumber":1239,"sourceCode":"            _finalize_write,\n            self.sink,\n            AsSingleton(init_result_coll),\n            AsIter(write_result_coll),\n            min_shards,\n            AsSingleton(pre_finalize_coll)).with_output_types(str))\n\n  def _apply_windowing(self, pcoll):\n    \"\"\"\n    Applies windowing to an unbounded PCollection based on the sink's\n    triggering frequency.\n    \"\"\"\n    use_fixed_windows = (\n        self.sink.triggering_frequency is not None and\n        self.sink.triggering_frequency > 0)\n\n    if isinstance(pcoll.windowing.windowfn, window.GlobalWindows):\n      if not use_fixed_windows:\n        raise ValueError(\n            'To write a GlobalWindow unbounded PCollection, '\n            'triggering_frequency must be set and be greater than 0')\n      return pcoll | 'ApplyFixedWindows' >> core.WindowInto(\n          window.FixedWindows(self.sink.triggering_frequency),\n          trigger=beam.transforms.trigger.AfterWatermark(),\n          accumulation_mode=beam.transforms.trigger.AccumulationMode.DISCARDING,\n          allowed_lateness=beam.utils.timestamp.Duration(seconds=0))\n\n    # Keep user-defined windowing unless triggering_frequency is specified.\n    if use_fixed_windows:\n      return pcoll | 'ApplyFixedWindows' >> core.WindowInto(\n          window.FixedWindows(self.sink.triggering_frequency),\n          trigger=beam.transforms.trigger.AfterWatermark(),\n          accumulation_mode=beam.transforms.trigger.AccumulationMode.DISCARDING,\n          allowed_lateness=beam.utils.timestamp.Duration(seconds=0))\n\n    return pcoll  # Keep original windowing\n","sourceCodeStart":1221,"sourceCodeEnd":1257,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/iobase.py#L1221-L1257","documentation":"When writing an unbounded (streaming) PCollection that is windowed into GlobalWindows, WriteToTransform._apply_windowing requires triggering_frequency, because GlobalWindows would otherwise buffer all records forever in sinks that must batch. The code fixes windowing into FixedWindows of size triggering_frequency with a discarding AfterWatermark trigger. Without it the write is unbounded and raises ValueError.","triggerScenarios":"Applying WriteToTransform (e.g. file-based sinks via WriteToFiles/Write) on a streaming PCollection whose windowfn is GlobalWindows while sink.triggering_frequency is None or <= 0.","commonSituations":"Streaming pipelines writing unbounded sources (Kafka/PubSub) to file sinks without setting triggering_frequency; copying batch write code into a streaming pipeline.","solutions":["Set triggering_frequency to a positive number of seconds on the sink, e.g. WriteToFiles(..., triggering_frequency=60).","Alternatively window the PCollection yourself into FixedWindows/SlidingWindows before the write.","If batch semantics were intended, ensure the PCollection is bounded (Read instead of a streaming source)."],"exampleFix":"# before\nbeam_data | WriteToFiles(path='/out')  # streaming, GlobalWindows\n# after\nbeam_data | WriteToFiles(path='/out', triggering_frequency=60)","handlingStrategy":"validation","validationCode":"if pcoll.is_streaming and isinstance(pcoll.windowing.windowfn, window.GlobalWindows) and not triggering_frequency:\n    raise ValueError('Set triggering_frequency > 0 for streaming GlobalWindow writes')","typeGuard":"def needs_fixed_windows(pcoll, triggering_frequency) -> bool:\n    from apache_beam.transforms import window\n    return pcoll.is_streaming and isinstance(pcoll.windowing.windowfn, window.GlobalWindows) and not (triggering_frequency and triggering_frequency > 0)","tryCatchPattern":"try:\n    out = pcoll | write_transform\nexcept ValueError as e:\n    if 'triggering_frequency' in str(e): log.error('Streaming write requires triggering_frequency')","preventionTips":["Always set triggering_frequency when writing streaming PCollections to file sinks","Prefer explicit WindowInto in streaming pipelines","Add pipeline-construction-time assertions for streaming writes"],"tags":["apache-beam","python","streaming","windowing","sink"],"backgroundTag":"invalid-config-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}