apache/beam · error · ValueError

Unconsumed error output for

Error message

Unconsumed error output for {identify_object(t)}. The output named {transform_name}.{error_output_name} must be used as an input to some other transform. See https://beam.apache.org/documentation/sdks/yaml-errors

What it means

Beam YAML requires that error outputs declared via error_handling.output actually be consumed — the error PCollection must feed some other transform. ensure_errors_consumed tracks (transform_id, output_name) pairs and raises this ValueError if a declared error output is never used as an input anywhere, preventing silent data loss of failed records.

Solutions

1. Add a transform whose input is <transform_name>.<error_output_name>, e.g. a WriteToJson or LogForTesting sink for errors.
2. Route the error output into a normalization/repair transform before sinking it.
3. Remove error_handling if the error branch is not actually wanted (errors then fail the pipeline instead).

Example fix

# before
- type: MapToFields
  name: parse
  config:
    error_handling:
      output: errors
# after
- type: MapToFields
  name: parse
  config:
    error_handling:
      output: errors
- type: WriteToJson
  input: {in: parse.errors}
  config:
    path: /tmp/errors.json

consumed = set()
for t in spec.get('transforms', []):
    for inp in t.get('input', {}).values():
        consumed.add(inp)
for t in spec.get('transforms', []):
    eh = t.get('config', t).get('error_handling')
    if eh:
        name = t.get('name', t['type'])
        if f'{name}.{eh["output"]}' not in consumed:
            print(f'warning: error output {name}.{eh["output"]} is unconsumed')

def error_output_consumed(spec, t):
    eh = t.get('config', t).get('error_handling')
    if not eh:
        return True
    ref = f"{t.get('name', t['type'])}.{eh['output']}"
    return any(ref in i.values() for i in [tr.get('input', {}) for tr in spec['transforms']])

try:
    spec = ensure_errors_consumed(spec)
except ValueError as e:
    if 'Unconsumed error output' in str(e):
        raise SystemExit(f'{e} — wire the error output into a sink or remove error_handling')
    raise

When it happens

Trigger: ensure_errors_consumed finds a (error_pcoll, t) pair in to_handle where error_pcoll is not in the consumed set — i.e. no transform's input references <that transform>.<error_output_name> and it is not a pipeline output.

Common situations: Adding error_handling to a transform but forgetting to wire the error output to a sink; deleting the error-handling branch during refactoring; typo in the output reference so the consumption check misses it.

Related errors

• Missing output in error_handling of
• At most one of --create_test and --fix_tests may be…
• Cannot convert element of type
• "Cannot specify 'callable' with 'path' and 'name' for…
• Chain at missing transforms property.

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

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_transform.py:1281

    consumed = set(
        scope.get_transform_id_and_output_name(output)
        for output in spec['output'].values())
    for t in spec['transforms']:
      config = t.get('config', t)
      if 'error_handling' in config:
        if 'output' not in config['error_handling']:
          raise ValueError(
              f'Missing output in error_handling of {identify_object(t)}')
        to_handle[t['__uuid__'], config['error_handling']['output']] = t
      for _, input in empty_if_explicitly_empty(t['input']).items():
        if input not in spec['input']:
          consumed.add(scope.get_transform_id_and_output_name(input))
    for error_pcoll, t in to_handle.items():
      if error_pcoll not in consumed:
        config = t.get('config', t)
        transform_name = t.get('name', t.get('type'))
        error_output_name = config['error_handling']['output']
        raise ValueError(
            f'Unconsumed error output for {identify_object(t)}. '
            f'The output named {transform_name}.{error_output_name} '
            'must be used as an input to some other transform. '
            'See https://beam.apache.org/documentation/sdks/yaml-errors')
  return spec


def lift_config(spec):
  if 'config' not in spec:
    common_params = 'name', 'type', 'input', 'output', 'transforms'
    return {
        'config': {
            k: v
            for (k, v) in spec.items() if k not in common_params
        },
        **{
            k: v
            for (k, v) in spec.items()  #

View on GitHub (pinned to 12126d8942)