apache/beam · error · ValueError

Missing input for transform

Error message

Missing input for transform: %s

What it means

Raised by validate_transform when an ASSIGN_WINDOWS (WindowInto) transform has no inputs. Window assignment is a 1-in/1-out primitive in the Beam model; a windowing node without an input is a malformed graph. Validation runs recursively from root_transform_ids via validate_pipeline_graph before execution.

Solutions

1. Wire exactly one input PCollection into the ASSIGN_WINDOWS transform (set transform_proto.inputs)
2. Rebuild the pipeline from Beam Python code (beam.WindowInto(...)) instead of editing protos
3. Fix custom translator code so WindowInto always consumes an input PCollection
4. Validate the pipeline graph with validate_pipeline_graph before submission in tooling

Example fix

// before
win.spec.ClearField('inputs')  # empty inputs -> raises
// after
win.spec.inputs['input'] = upstream_pc_id
assert len(win.spec.inputs) == 1

def check_window_has_input(t: beam_runner_api_pb2.PTransform) -> bool:
    return (t.spec.urn != common_urns.primitives.ASSIGN_WINDOWS.urn
            or bool(t.inputs))

def window_is_wired(t) -> bool:
    return t.spec.urn != 'beam:transform:assign_windows:v1' or len(t.inputs) > 0

try:
    validate_pipeline_graph(pipeline_proto)
except ValueError as e:
    if 'Missing input for transform' in str(e):
        rewire_orphan_transforms(pipeline_proto)
    else:
        raise

When it happens

Trigger: A pipeline proto containing an ASSIGN_WINDOWS transform with an empty transform_proto.inputs map — from hand-constructed protos, buggy translators, or externally modified graphs.

Common situations: Custom graph-building tooling that drops windowing input edges; pipeline fragment merging that lost the input PCollection reference; hand-edited serialized pipeline JSON.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• accumulation_mode must be provided for non-trivial triggers
• assign_context.window should not be None. This might be due…
• Bad coder for input of
• Bad coder for output of
• Default values are not yet supported in CombineGlobally()…

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/pipeline_utils.py:122

      output_values_coder = pipeline_proto.components.coders[
          output_values_coder_id]
      if output_values_coder.spec.urn != common_urns.coders.ITERABLE.urn:
        raise ValueError(
            "Output value coder %s for transform %s must be an iterable "
            "coder, but uses URN %s" % (
                output_values_coder_id,
                transform_id,
                output_values_coder.spec.urn))
      input_value_coder_id = input_coder.component_coder_ids[1]
      output_value_coder_id = output_values_coder.component_coder_ids[0]
      if output_value_coder_id != input_value_coder_id:
        raise ValueError(
            "Input value coder %s does not match output value coder %s for "
            "transform %s" %
            (input_value_coder_id, output_value_coder_id, transform_id))
    elif transform_proto.spec.urn == common_urns.primitives.ASSIGN_WINDOWS.urn:
      if not transform_proto.inputs:
        raise ValueError("Missing input for transform: %s" % transform_proto)
    elif transform_proto.spec.urn == common_urns.primitives.PAR_DO.urn:
      if not transform_proto.inputs:
        raise ValueError("Missing input for transform: %s" % transform_proto)

    for t in transform_proto.subtransforms:
      validate_transform(t)

  for t in pipeline_proto.root_transform_ids:
    validate_transform(t)


def _dep_key(dep):
  if dep.type_urn == common_urns.artifact_types.FILE.urn:
    payload = beam_runner_api_pb2.ArtifactFilePayload.FromString(
        dep.type_payload)
    if payload.sha256:
      type_info = 'sha256', payload.sha256
    else:

View on GitHub (pinned to 12126d8942)