apache/beam · error · ValueError

Missing inputs for transform at

Error message

Missing inputs for transform at {identify_object(spec)}

What it means

Most transforms require input PCollections. If a transform receives no inputs (`input_pcolls` empty), the `input` field is not explicitly empty, and its provider declares `requires_inputs(type, config)` true, create_ptransform raises this ValueError at the spec's location.

Solutions

  1. Add an `input:` field referencing an existing source transform or pipeline input.
  2. Fix indentation so `input:` is a top-level key of the transform spec.
  3. If the transform is intentionally a root, use a source type (e.g. ReadFromBigQuery) that does not require inputs.
  4. Trace the chain to ensure the referenced upstream transform still exists.

Example fix

# before
- name: filter_rows
  type: Filter
  config:
    keep: 'x > 0'
# after
- name: filter_rows
  type: Filter
  input: read_rows
  config:
    keep: 'x > 0'
Defensive patterns

Strategy: validation

Validate before calling

SOURCES = {'ReadFromBigQuery', 'ReadFromKafka', 'ReadFromPubSub'}
for t in pipeline['transforms']:
    if t['type'] not in SOURCES and 'input' not in t:
        raise ValueError(f'{t.get("name")}: non-source transform needs input:')

Type guard

def has_input(spec):
    return 'input' in spec and spec['input'] is not None

Try / catch

try:
    run_pipeline(spec)
except ValueError as e:
    if 'Missing inputs' in str(e):
        raise UserPipelineError(f'{spec.get("name")}: add an input: reference') from e

Prevention

When it happens

Trigger: Declaring a consumer transform (e.g. MapToFields, Filter, Write) without an `input:` field, or with an input that resolves to nothing; source-type transforms misconfigured so they are treated as consumers.

Common situations: Forgetting the `input:` key on non-source transforms; indentation placing `input` inside `config`; deleting an upstream transform while leaving its consumer; broken input references.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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

Appendix: source

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

    providers_by_input = {k: v for k, v in self.input_providers.items()}
    input_providers = [
        providers_by_input[pcoll] for pcoll in input_pcolls
        if pcoll in providers_by_input
    ]
    provider = self.best_provider(spec, input_providers)
    extra_dependencies, spec = extract_extra_dependencies(spec)
    if extra_dependencies:
      provider = provider.with_extra_dependencies(frozenset(extra_dependencies))

    config = SafeLineLoader.strip_metadata(spec.get('config', {}))
    if not isinstance(config, dict):
      raise ValueError(
          'Config for transform at %s must be a mapping.' %
          identify_object(spec))

    if (not input_pcolls and not is_explicitly_empty(spec.get('input', {})) and
        provider.requires_inputs(spec['type'], config)):
      raise ValueError(
          f'Missing inputs for transform at {identify_object(spec)}')

    try:
      if spec['type'].endswith('-generic'):
        # Centralize the validation rather than require every implementation
        # to do it.
        validate_generic_expressions(
            spec['type'].rsplit('-', 1)[0], config, input_pcolls)

      # pylint: disable=undefined-loop-variable
      ptransform = maybe_with_resource_hints(
          provider.create_transform(
              spec['type'],
              config,
              lambda config, input_pcolls=input_pcolls: self.create_ptransform(
                  config, input_pcolls)))
      # TODO(robertwb): Should we have a better API for adding annotations
      # than this?

View on GitHub (pinned to 12126d8942)