{"record":{"id":"35df4dcce49d6001","repo":"apache/beam","slug":"groupbykey-cannot-be-applied-to-an-unbounded-pcollection","errorCode":null,"errorMessage":"GroupByKey cannot be applied to an unbounded PCollection with global windowing and a default trigger","messagePattern":"GroupByKey cannot be applied to an unbounded PCollection with global windowing and a default trigger","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/core.py","lineNumber":3517,"sourceCode":"      return (pcoll | \"Group by encrypted key\" >> GroupByEncryptedKey(secret))\n\n    from apache_beam.transforms.trigger import DataLossReason\n    from apache_beam.transforms.trigger import DefaultTrigger\n    windowing = pcoll.windowing\n    trigger = windowing.triggerfn\n    if not pcoll.is_bounded and isinstance(\n        windowing.windowfn, GlobalWindows) and isinstance(trigger,\n                                                          DefaultTrigger):\n      if pcoll.pipeline.allow_unsafe_triggers:\n        # TODO(BEAM-9487) Change comment for Beam 2.33\n        _LOGGER.warning(\n            '%s: PCollection passed to GroupByKey is unbounded, has a global '\n            'window, and uses a default trigger. This is being allowed '\n            'because --allow_unsafe_triggers is set, but it may prevent '\n            'data from making it through the pipeline.',\n            self.label)\n      else:\n        raise ValueError(\n            'GroupByKey cannot be applied to an unbounded ' +\n            'PCollection with global windowing and a default trigger')\n\n    unsafe_reason = trigger.may_lose_data(windowing)\n    if unsafe_reason != DataLossReason.NO_POTENTIAL_LOSS:\n      reason_msg = str(unsafe_reason).replace('DataLossReason.', '')\n      if pcoll.pipeline.allow_unsafe_triggers:\n        _LOGGER.warning(\n            '%s: Unsafe trigger `%s` detected (reason: %s). This is '\n            'being allowed because --allow_unsafe_triggers is set. This could '\n            'lead to missing or incomplete groups.',\n            self.label,\n            trigger,\n            reason_msg)\n      else:\n        msg = '{}: Unsafe trigger: `{}` may lose data. '.format(\n            self.label, trigger)\n        msg += 'Reason: {}. '.format(reason_msg)","sourceCodeStart":3499,"sourceCodeEnd":3535,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/core.py#L3499-L3535","documentation":"GroupByKey on an unbounded (streaming) PCollection with GlobalWindows and the default trigger cannot complete, because the global window never closes and the default trigger never fires. Beam raises this ValueError unless --allow_unsafe_triggers is set (in which case it only warns).","triggerScenarios":"Applying beam.GroupByKey() to a streaming pipeline's PCollection where windowing is GlobalWindows and the trigger is the default (i.e. windowing was not customized) — typically in a streaming runner like Flink/Dataflow with unbounded sources.","commonSituations":"Developers writing a pipeline meant for batch and running it in streaming mode without setting windows/triggers; or streaming jobs that read from Pub/Sub/Kafka without applying window transforms before grouping.","solutions":["Apply a non-global windowing before the GroupByKey, e.g. pcoll | beam.WindowInto(beam.window.FixedWindows(60))","Set a non-default trigger compatible with the windowing (e.g. default trigger replaced by one that fires repeatedly)","Pass --allow_unsafe_triggers only as a temporary workaround, knowing data may never pass through","Switch the pipeline to bounded mode if streaming semantics were not intended"],"exampleFix":"// before\npcoll | beam.GroupByKey()\n// after\npcoll | beam.WindowInto(beam.window.FixedWindows(60)) | beam.GroupByKey()","handlingStrategy":"validation","validationCode":"from apache_beam import window as bw\nif is_streaming and pcoll.windowing.windowfn == bw.GlobalWindows() and pcoll.windowing.is_default():\n    pcoll = pcoll | beam.WindowInto(bw.FixedWindows(60))","typeGuard":"def gbk_safe_windowing(windowing) -> bool:\n    from apache_beam.transforms import window\n    return not (windowing.windowfn == window.GlobalWindows() and windowing.triggerfn == window.GloballyWindows().__class__ and windowing.is_default()) if hasattr(windowing, 'is_default') else type(windowing.windowfn).__name__ != 'GlobalWindows'","tryCatchPattern":"try:\n    expanded = pcoll | beam.GroupByKey()\nexcept ValueError as e:\n    if 'unbounded PCollection with global windowing' in str(e):\n        expanded = pcoll | beam.WindowInto(beam.window.FixedWindows(60)) | beam.GroupByKey()\n    else:\n        raise","preventionTips":["In streaming pipelines, always apply WindowInto before any grouping","Avoid reading unbounded sources without windowing configuration","Detect bounded vs unbounded at pipeline start and branch your windowing strategy"],"tags":["python","apache-beam","groupbykey","streaming","windowing"],"backgroundTag":"invalid-state-transition","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"}