apache/beam · error · ValueError

GroupByKey cannot be applied to an unbounded PCollection wit

Error message

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

What it means

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).

Source

Thrown at sdks/python/apache_beam/transforms/core.py:3517

      return (pcoll | "Group by encrypted key" >> GroupByEncryptedKey(secret))

    from apache_beam.transforms.trigger import DataLossReason
    from apache_beam.transforms.trigger import DefaultTrigger
    windowing = pcoll.windowing
    trigger = windowing.triggerfn
    if not pcoll.is_bounded and isinstance(
        windowing.windowfn, GlobalWindows) and isinstance(trigger,
                                                          DefaultTrigger):
      if pcoll.pipeline.allow_unsafe_triggers:
        # TODO(BEAM-9487) Change comment for Beam 2.33
        _LOGGER.warning(
            '%s: PCollection passed to GroupByKey is unbounded, has a global '
            'window, and uses a default trigger. This is being allowed '
            'because --allow_unsafe_triggers is set, but it may prevent '
            'data from making it through the pipeline.',
            self.label)
      else:
        raise ValueError(
            'GroupByKey cannot be applied to an unbounded ' +
            'PCollection with global windowing and a default trigger')

    unsafe_reason = trigger.may_lose_data(windowing)
    if unsafe_reason != DataLossReason.NO_POTENTIAL_LOSS:
      reason_msg = str(unsafe_reason).replace('DataLossReason.', '')
      if pcoll.pipeline.allow_unsafe_triggers:
        _LOGGER.warning(
            '%s: Unsafe trigger `%s` detected (reason: %s). This is '
            'being allowed because --allow_unsafe_triggers is set. This could '
            'lead to missing or incomplete groups.',
            self.label,
            trigger,
            reason_msg)
      else:
        msg = '{}: Unsafe trigger: `{}` may lose data. '.format(
            self.label, trigger)
        msg += 'Reason: {}. '.format(reason_msg)

View on GitHub (pinned to 12126d8942)

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

Example fix

// before
pcoll | beam.GroupByKey()
// after
pcoll | beam.WindowInto(beam.window.FixedWindows(60)) | beam.GroupByKey()

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam import window as bw
if is_streaming and pcoll.windowing.windowfn == bw.GlobalWindows() and pcoll.windowing.is_default():
    pcoll = pcoll | beam.WindowInto(bw.FixedWindows(60))

Type guard

def gbk_safe_windowing(windowing) -> bool:
    from apache_beam.transforms import window
    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'

Try / catch

try:
    expanded = pcoll | beam.GroupByKey()
except ValueError as e:
    if 'unbounded PCollection with global windowing' in str(e):
        expanded = pcoll | beam.WindowInto(beam.window.FixedWindows(60)) | beam.GroupByKey()
    else:
        raise

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/35df4dcce49d6001. Report an issue: GitHub.