apache/beam · error · ValueError

Unexpected number of outputs

Error message

Unexpected number of outputs: %s

What it means

Raised by validate_transform: a GROUP_BY_KEY transform must have exactly one output, and the proto declares another count. GBK in the Beam model is a 1-in/1-out primitive; any other shape means the pipeline graph is malformed. The check runs during validate_pipeline_graph before a runner executes the pipeline.

Solutions

1. Fix the failing transform so it has exactly one entry in transform_proto.outputs
2. Rebuild the pipeline from Beam code rather than constructing/altering protos manually
3. If fanning out results, insert explicit ParDo/flatten transforms after GBK instead of multiple GBK outputs
4. Align SDK/runner versions and re-serialize the pipeline

Example fix

// before
gbk.spec.outputs['main'] = pc1
gbk.spec.outputs['extra'] = pc2  # 2 outputs -> raises
// after
del gbk.spec.outputs['extra']
assert len(gbk.spec.outputs) == 1

def check_gbk_outputs(t: beam_runner_api_pb2.PTransform) -> bool:
    return (t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn
            or len(t.outputs) == 1)

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

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

When it happens

Trigger: A GROUP_BY_KEY transform whose transform_proto.outputs map has 0 or multiple entries — from hand-built protos, buggy custom translators, or externally modified pipeline graphs.

Common situations: Custom runner or graph-rewrite tooling that fans out GBK output; serialized pipelines edited by external tools; portability graph conversion bugs between SDKs.

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 inputs
• 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/fc0a4758cb48cbcb. Report an issue: GitHub.

Appendix: source

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

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]
      output_values_coder = pipeline_proto.components.coders[
          output_values_coder_id]

View on GitHub (pinned to 12126d8942)