apache/beam · error · ValueError

Explicit output of the chain transform is not an output of…

Error message

Explicit output {identify_object(value)} of the chain transform is not an output of the last transform.

What it means

In Beam YAML pipelines, a 'chain' transform is expanded into a composite whose external output must be produced by the last transform in the chain. If the spec has an explicit 'output' mapping that references a value not produced by the last transform, chain_as_composite rejects it so the pipeline cannot silently drop or misroute outputs.

Solutions

1. Update the explicit output mapping so each value names an output of the last transform in the chain (or just the transform's default output name).
2. If you need an intermediate result, split the chain or move that transform to be last.
3. Remove the explicit 'output' section to let the chain default to the last transform's output.

Example fix

# before
- type: chain
  transforms:
    - type: ReadFromText
      name: Read
    - type: WriteToJson
      name: Write
  output: {result: Read}
# after
- type: chain
  transforms:
    - type: ReadFromText
      name: Read
    - type: WriteToJson
      name: Write
  output: {result: Write}

def check_chain_outputs(spec):
    transforms = spec.get('transforms', [])
    if not transforms:
        return
    last = transforms[-1]
    for key, value in (spec.get('output') or {}).items():
        owner = value.split('.')[0] if '.' in value else last.get('name', last['type'])
        if owner != last.get('name', last['type']):
            raise ValueError(f"output {key}->{value} not produced by last transform")

def output_belongs_to_last(spec):
    last_name = spec['transforms'][-1].get('name', spec['transforms'][-1]['type'])
    return all(v.split('.')[0] == last_name for v in spec.get('output', {}).values())

try:
    spec = chain_as_composite(spec)
except ValueError as e:
    if 'not an output of the last transform' in str(e):
        spec['output'] = {k: v.split('.')[-1] for k, v in spec['output'].items()}
    else:
        raise

When it happens

Trigger: Calling chain_as_composite (directly or via ExpandChainTransform/preprocessing) on a spec where composite_spec['output'] maps a key to a value for which is_not_output_of_last_transform(new_transforms, value) is True — e.g. output refers to an intermediate transform's output or a nonexistent output name instead of the final transform's output.

Common situations: Hand-written YAML where the top-level output key points at an earlier step in a chain (e.g. ReadToWrite chain exposing 'Read' output); renaming the last transform but forgetting to update output references; copy-pasting output blocks between chains.

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

• At most one of --create_test and --fix_tests may be…
• Bad coder for input of
• Bad coder for output of
• Cannot convert element of type
• "Cannot specify 'callable' with 'path' and 'name' for…

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

Appendix: source

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

        transform['input'] = composite_spec['input']
      elif is_empty(composite_spec['input']):
        del composite_spec['input']
      else:
        transform['input'] = {
            key: key
            for key in composite_spec['input'].keys()
        }
    else:
      transform['input'] = new_transforms[-1]['__uuid__']
    new_transforms.append(transform)
  new_transforms.extend(spec.get('extra_transforms', []))
  composite_spec['transforms'] = new_transforms

  last_transform = new_transforms[-1]['__uuid__']
  if has_explicit_outputs:
    for (key, value) in composite_spec['output'].items():
      if is_not_output_of_last_transform(new_transforms, value):
        raise ValueError(
            f"Explicit output {identify_object(value)} of the chain transform"
            f" is not an output of the last transform.")

    composite_spec['output'] = {
        key: f'{last_transform}.{value}'
        for (key, value) in composite_spec['output'].items()
    }
  else:
    composite_spec['output'] = {'__implicit_outputs__': last_transform}
  if 'name' not in composite_spec:
    composite_spec['name'] = 'Chain'
  composite_spec['type'] = 'composite'
  return composite_spec


def preprocess_chain(spec):
  if spec['type'] == 'chain':
    return chain_as_composite(spec)

View on GitHub (pinned to 12126d8942)