apache/beam · error · ValueError

: Unsafe trigger: ` ` may lose data. Reason: . This can be…

Error message

{}: Unsafe trigger: `{}` may lose data. Reason: {}. This can be overriden with the --allow_unsafe_triggers flag.

What it means

Beam's GroupByKey validates that the configured trigger does not potentially discard data (may_lose_data returns NO_POTENTIAL_LOSS). If the trigger may lose data, this ValueError is raised naming the transform label, the trigger, and the specific DataLossReason, and pointing to the --allow_unsafe_triggers escape hatch.

Solutions

Redesign the trigger to be safe, e.g. use Repeatedly/AfterWatermark with allowed lateness and with_accumulation_mode(ACCUMULATING)
Increase allowed_lateness in WindowInto so late data is not discarded
Pass --allow_unsafe_triggers to override, only after accepting possible data loss
Log the reason_msg (from DataLossReason) to understand exactly which data can be lost

Example fix

// before
beam.WindowInto(FixedWindows(60), trigger=AfterCount(1))
// after
beam.WindowInto(FixedWindows(60),
  trigger=Repeatedly(AfterWatermark(late=AfterCount(1))),
  accumulation_mode=beam.trigger.AccumulationMode.ACCUMULATING,
  allowed_lateness=300)

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.transforms.trigger import DataLossReason
unsafe = trigger.may_lose_data(windowing)
assert unsafe == DataLossReason.NO_POTENTIAL_LOSS, f'trigger may lose data: {unsafe}'

Type guard

def is_safe_trigger(trigger, windowing) -> bool:
    from apache_beam.transforms.trigger import DataLossReason
    return trigger.may_lose_data(windowing) == DataLossReason.NO_POTENTIAL_LOSS

Try / catch

try:
    expanded = pcoll | beam.GroupByKey()
except ValueError as e:
    if 'Unsafe trigger' in str(e):
        expanded = (pcoll | beam.WindowInto(
            beam.window.FixedWindows(60),
            allowed_lateness=300,
            accumulation_mode=beam.trigger.AccumulationMode.ACCUMULATING)
            | beam.GroupByKey())
    else:
        raise

Prevention

Check trigger.may_lose_data(windowing) when designing custom triggers
Prefer Repeatedly/AfterWatermark with allowed lateness over one-shot triggers
Treat --allow_unsafe_triggers as a last-resort diagnostic flag, not a production setting

When it happens

Trigger: Applying GroupByKey on a windowed streaming PCollection whose WindowInto uses a trigger such as a one-shot AfterCount or AfterWatermark without allowed lateness/accumulation that Beam deems unsafe — any trigger where may_lose_data() != NO_POTENTIAL_LOSS.

Common situations: Custom triggers built with beam.trigger.* (e.g. AfterProcessingTime with unbounded lateness, element-count triggers) combined with grouping; a runner upgrade starts surfacing the check on previously-accepted pipelines.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

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

Appendix: source

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

            '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)
        msg += 'This can be overriden with the --allow_unsafe_triggers flag.'
        raise ValueError(msg)

    return pvalue.PCollection.from_(pcoll)

  def infer_output_type(self, input_type):
    key_type, value_type = (typehints.typehints.coerce_to_kv_type(
        input_type).tuple_types)
    return typehints.KV[key_type, typehints.Iterable[value_type]]

  def to_runner_api_parameter(self, unused_context):
    # type: (PipelineContext) -> tuple[str, typing.Optional[typing.Union[message.Message, bytes, str]]]
    # if we're containing a GroupByEncryptedKey, don't allow runners to
    # recognize this transform as a GBEK so that it doesn't get replaced.
    if self._replaced_by_gbek:
      return super().to_runner_api_parameter(unused_context)
    return common_urns.primitives.GROUP_BY_KEY.urn, None

  @staticmethod
  @PTransform.register_urn(common_urns.primitives.GROUP_BY_KEY.urn, None)

View on GitHub (pinned to 12126d8942)