{"record":{"id":"74b242e8f2e48919","repo":"apache/beam","slug":"the-number-of-failing-elements-within-the-window-r-exceeded","errorCode":null,"errorMessage":"The number of failing elements within the window %r exceeded threshold: %s / %s = %s > %s","messagePattern":"The number of failing elements within the window %r exceeded threshold: (.+?) / (.+?) = (.+?) > (.+?)","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/core.py","lineNumber":2435,"sourceCode":"          if self._threshold_windowing:\n            return pcoll | WindowInto(self._threshold_windowing)\n          else:\n            return pcoll\n\n      # Map(lambda) produces a label formatted like this, but it cannot be\n      # changed without breaking update compat. Here, we pin to the transform\n      # name used in the 2.68 release to avoid breaking changes when the line\n      # number changes. Context: https://github.com/apache/beam/pull/36381\n      input_count_view = pcoll | 'CountTotal' >> (\n          MaybeWindow() | \"Map(<lambda at core.py:2346>)\" >> Map(lambda _: 1)\n          | CombineGlobally(sum).as_singleton_view())\n      bad_count_pcoll = result[self._dead_letter_tag] | 'CountBad' >> (\n          MaybeWindow() | \"Map(<lambda at core.py:2349>)\" >> Map(lambda _: 1)\n          | CombineGlobally(sum).without_defaults())\n\n      def check_threshold(bad, total, threshold, window=DoFn.WindowParam):\n        if bad > total * threshold:\n          raise ValueError(\n              'The number of failing elements within the window %r '\n              'exceeded threshold: %s / %s = %s > %s' %\n              (window, bad, total, bad / total, threshold))\n\n      _ = bad_count_pcoll | Map(\n          check_threshold, input_count_view, self._threshold)\n\n    if self._error_handler:\n      self._error_handler.add_error_pcollection(result[self._dead_letter_tag])\n      if self._extra_tags is not None:\n        return result\n      return result[self._main_tag]\n    else:\n      return result\n\n  def expand_2_72_0(self, pcoll):\n    \"\"\"Pre-2.73.0 behavior: manual element_type override, no with_output_types.\n    \"\"\"","sourceCodeStart":2417,"sourceCodeEnd":2453,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/core.py#L2417-L2453","documentation":"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).","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\n | Map(fn).with_exception_handling(dead_letter_tag='bad', threshold=0.01)\n// after\n | Map(fn).with_exception_handling(dead_letter_tag='bad', threshold=0.1)","handlingStrategy":"validation","validationCode":"bad_ratio = bad_count / total_count\nassert bad_ratio <= threshold, f'failure ratio {bad_ratio} exceeds threshold {threshold}'","typeGuard":null,"tryCatchPattern":"result = (pcoll | beam.Map(risky_fn).with_exception_handling(dead_letter_tag='bad', threshold=t))\n# consume 'result[\"bad\"]' separately; check_threshold raising ValueError surfaces at runtime:\ntry:\n    run_pipeline()\nexcept ValueError as e:\n    if 'exceeded threshold' in str(e):\n        alert_data_quality_team()\n    else:\n        raise","preventionTips":["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"],"tags":["python","apache-beam","data-quality","threshold"],"backgroundTag":"value-out-of-range","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}