apache/beam · error · ValueError

Unexpected number of inputs

Error message

Unexpected number of inputs: %s

What it means

Raised by validate_transform while validating a pipeline's model proto: a GROUP_BY_KEY (GroupByKey) transform must have exactly one input, and the pipeline's proto declares a different number. Beam validates the serialized pipeline graph before handing it to a runner. This indicates a malformed or hand-assembled pipeline proto rather than a normal user-level Beam error.

Solutions

1. Inspect the transform_proto.inputs of the failing GBK transform (ID is in the message) and correct it to exactly one input PCollection
2. Regenerate the pipeline from Pipeline/Beam Python code instead of editing or constructing protos by hand
3. Verify all SDK/runner components use compatible Beam versions; recompile/re-serialize the pipeline proto
4. If writing custom translators, ensure your GBK replacement has exactly one input map entry

Example fix

// before
gbk.spec.inputs['main'] = pc_in
gbk.spec.inputs['side'] = pc_extra  # 2 inputs -> raises
// after
del gbk.spec.inputs['side']  # exactly one input
assert len(gbk.spec.inputs) == 1

from apache_beam.portability.api import beam_runner_api_pb2
def check_gbk_inputs(t: beam_runner_api_pb2.PTransform) -> bool:
    return (t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn
            or len(t.inputs) == 1)

def is_wellformed_gbk(t) -> bool:
    return t.spec.urn != 'beam:transform:group_by_key:v1' or len(t.inputs) == 1

try:
    validate_pipeline_graph(pipeline_proto)
except ValueError as e:
    if 'Unexpected number of inputs' in str(e):
        rebuild_pipeline_graph(pipeline_proto)
    else:
        raise

When it happens

Trigger: Submitting a pipeline whose proto contains a GROUP_BY_KEY transform with 0 or 2+ entries in transform_proto.inputs — typically from manually constructed protos, cross-language portability graphs, or corrupted/older serialized pipeline JSON passed through validate_pipeline_graph.

Common situations: Building Pipeline protos programmatically or in tooling; round-tripping pipeline graphs through external storage; mixing SDK versions where a runner or translator produced a nonstandard GBK node.

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

• Unexpected number of outputs
• Bad coder for input of
• Bad coder for output of
• Explicit output of the chain transform is not an output of…
• Input key coder does not match output key coder for…

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

Appendix: source

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

  return GroupByKeyInputVisitor(deterministic_key_coders)


def validate_pipeline_graph(pipeline_proto):
  """Ensures this is a correctly constructed Beam pipeline.
  """
  def get_coder(pcoll_id):
    return pipeline_proto.components.coders[
        pipeline_proto.components.pcollections[pcoll_id].coder_id]

  def validate_transform(transform_id):
    transform_proto = pipeline_proto.components.transforms[transform_id]

    # Currently the only validation we perform is that GBK operations have
    # their coders set properly.
    if transform_proto.spec.urn == common_urns.primitives.GROUP_BY_KEY.urn:
      if len(transform_proto.inputs) != 1:
        raise ValueError("Unexpected number of inputs: %s" % transform_proto)
      if len(transform_proto.outputs) != 1:
        raise ValueError("Unexpected number of outputs: %s" % transform_proto)
      input_coder = get_coder(next(iter(transform_proto.inputs.values())))
      output_coder = get_coder(next(iter(transform_proto.outputs.values())))
      if input_coder.spec.urn != common_urns.coders.KV.urn:
        raise ValueError(
            "Bad coder for input of %s: %s" % (transform_id, input_coder))
      if output_coder.spec.urn != common_urns.coders.KV.urn:
        raise ValueError(
            "Bad coder for output of %s: %s" % (transform_id, output_coder))
      input_key_coder_id = input_coder.component_coder_ids[0]
      output_key_coder_id = output_coder.component_coder_ids[0]
      if input_key_coder_id != output_key_coder_id:
        raise ValueError(
            "Input key coder %s does not match output key coder %s for "
            "transform %s" %
            (input_key_coder_id, output_key_coder_id, transform_id))
      output_values_coder_id = output_coder.component_coder_ids[1]

View on GitHub (pinned to 12126d8942)