apache/beam · error · ValueError

Transform is part of a chain. Cannot define explicit inputs…

Error message

Transform {identify_object(transform)} is part of a chain. Cannot define explicit inputs on chain pipeline

What it means

Inside a chain transform, inputs and outputs flow implicitly from one step to the next, so individual steps may not declare their own 'input'/'output' (except the first step may set an explicitly empty input, e.g. a source). chain_as_composite raises this ValueError when any chain step defines explicit io that isn't that allowed exception.

Solutions

  1. Remove 'input'/'output' keys from chain steps; chaining wires them automatically.
  2. If explicit wiring is needed, convert the chain to a 'composite' transform where explicit inputs/outputs are allowed.
  3. For custom output names, use the chain's top-level 'output' override instead of per-step outputs.
  4. If a step must start from an empty source, keep it as the first step and set its input to explicitly empty ({}), which is permitted.

Example fix

// before
- type: chain
  transforms:
    - type: MapToFields
      input: source
      config: {id: element.id}
// after
- type: chain
  input: source
  transforms:
    - type: MapToFields
      config: {id: element.id}
Defensive patterns

Strategy: validation

Validate before calling

def check_chain_io(spec):
    for t in spec.get('transforms', []):
        if 'input' in t or 'output' in t:
            raise ValueError(f"Chain step {t.get('name', t.get('type'))} must not declare explicit input/output")

Type guard

def chain_steps_are_implicit(spec) -> bool:
    return all('input' not in t and 'output' not in t for t in spec.get('transforms', []))

Try / catch

try:
    expand_transform(spec, scope)
except ValueError as e:
    if 'Cannot define explicit inputs on chain pipeline' in str(e):
        print('Remove input/output keys from chain steps or convert to composite')
    else:
        raise

Prevention

When it happens

Trigger: A transforms entry within a chain spec containing 'input:' or 'output:' keys — e.g. an intermediate step with input: previous_step, or a step with output: my_tag — anything beyond the first step's explicitly-empty input.

Common situations: Converting a composite pipeline to a chain and leaving explicit input/output references in the steps; adding a named output tag inside a chain to branch results; copy-pasting steps from composite sections into chain sections.

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/4a163539a2459e99. Report an issue: GitHub.

Appendix: source

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

  # A chain is simply a composite transform where all inputs and outputs
  # are implicit.
  spec = normalize_source_sink(spec)
  if 'transforms' not in spec:
    raise TypeError(
        f"Chain at {identify_object(spec)} missing transforms property.")
  has_explicit_outputs = 'output' in spec
  composite_spec = dict(normalize_inputs_outputs(tag_explicit_inputs(spec)))
  new_transforms = []
  for ix, transform in enumerate(composite_spec['transforms']):
    transform = dict(transform)
    if any(io in transform for io in ('input', 'output')):
      if (ix == 0 and 'input' in transform and 'output' not in transform and
          is_explicitly_empty(transform['input'])):
        # This is OK as source clause sets an explicitly empty input.
        pass
      else:
        raise ValueError(
            f'Transform {identify_object(transform)} is part of a chain. '
            'Cannot define explicit inputs on chain pipeline')
    if ix == 0:
      if is_explicitly_empty(transform.get('input', None)):
        pass
      elif is_explicitly_empty(composite_spec['input']):
        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', []))

View on GitHub (pinned to 12126d8942)