apache/beam · error · ValueError

The number of failing elements within the window %r…

Error message

The number of failing elements within the window %r exceeded threshold: %s / %s = %s > %s

What it means

This ValueError comes from a user-supplied or built-in threshold check appended after a dead-letter (exception-handling) ParDo. It counts elements routed to the dead-letter tag per window and fails the pipeline when the fraction of failing elements exceeds the configured threshold (bad / total > threshold).

Solutions

Inspect the dead-letter output collection to see which elements fail and why
Fix or filter the bad data upstream (add validation / fallback parsing)
Raise the threshold parameter if the current failure rate is acceptable
Route failing elements to a separate durable sink instead of failing the whole pipeline

Example fix

// before
 | Map(fn).with_exception_handling(dead_letter_tag='bad', threshold=0.01)
// after
 | Map(fn).with_exception_handling(dead_letter_tag='bad', threshold=0.1)

Defensive patterns

Strategy: validation

Validate before calling

bad_ratio = bad_count / total_count
assert bad_ratio <= threshold, f'failure ratio {bad_ratio} exceeds threshold {threshold}'

Try / catch

result = (pcoll | beam.Map(risky_fn).with_exception_handling(dead_letter_tag='bad', threshold=t))
# consume 'result["bad"]' separately; check_threshold raising ValueError surfaces at runtime:
try:
    run_pipeline()
except ValueError as e:
    if 'exceeded threshold' in str(e):
        alert_data_quality_team()
    else:
        raise

Prevention

Monitor dead-letter counts continuously instead of only at threshold failure
Set thresholds from measured baseline failure rates, not guesses
Keep a sample of failing elements in the dead-letter output for diagnosis

When it happens

Trigger: Applying a ParDo with with_exception_handling(..., threshold=X) (or wiring the dead-letter counting transform manually) and, within some window, more than X fraction of elements raise exceptions: e.g. threshold=0.1 and 2 of 10 elements fail in one window.

Common situations: Pipeline data quality gates: a downstream API starts returning 4xx for a subset of records, schema drift increases parse failures, or a transient outage pushes the failure ratio above the configured limit, aborting the pipeline.

Understand the failure class

Background: "value must be between 0 and 1" / "out of range" / "must not be negative" errors: fixing range-validation failures across open-source libraries — this error's family across 42 libraries.

Related errors

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

Appendix: source

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

          if self._threshold_windowing:
            return pcoll | WindowInto(self._threshold_windowing)
          else:
            return pcoll

      # Map(lambda) produces a label formatted like this, but it cannot be
      # changed without breaking update compat. Here, we pin to the transform
      # name used in the 2.68 release to avoid breaking changes when the line
      # number changes. Context: https://github.com/apache/beam/pull/36381
      input_count_view = pcoll | 'CountTotal' >> (
          MaybeWindow() | "Map(<lambda at core.py:2346>)" >> Map(lambda _: 1)
          | CombineGlobally(sum).as_singleton_view())
      bad_count_pcoll = result[self._dead_letter_tag] | 'CountBad' >> (
          MaybeWindow() | "Map(<lambda at core.py:2349>)" >> Map(lambda _: 1)
          | CombineGlobally(sum).without_defaults())

      def check_threshold(bad, total, threshold, window=DoFn.WindowParam):
        if bad > total * threshold:
          raise ValueError(
              'The number of failing elements within the window %r '
              'exceeded threshold: %s / %s = %s > %s' %
              (window, bad, total, bad / total, threshold))

      _ = bad_count_pcoll | Map(
          check_threshold, input_count_view, self._threshold)

    if self._error_handler:
      self._error_handler.add_error_pcollection(result[self._dead_letter_tag])
      if self._extra_tags is not None:
        return result
      return result[self._main_tag]
    else:
      return result

  def expand_2_72_0(self, pcoll):
    """Pre-2.73.0 behavior: manual element_type override, no with_output_types.
    """

View on GitHub (pinned to 12126d8942)