{"record":{"id":"fc0a4758cb48cbcb","repo":"apache/beam","slug":"unexpected-number-of-outputs-s","errorCode":null,"errorMessage":"Unexpected number of outputs: %s","messagePattern":"Unexpected number of outputs: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/pipeline_utils.py","lineNumber":87,"sourceCode":"\n\ndef validate_pipeline_graph(pipeline_proto):\n  \"\"\"Ensures this is a correctly constructed Beam pipeline.\n  \"\"\"\n  def get_coder(pcoll_id):\n    return pipeline_proto.components.coders[\n        pipeline_proto.components.pcollections[pcoll_id].coder_id]\n\n  def validate_transform(transform_id):\n    transform_proto = pipeline_proto.components.transforms[transform_id]\n\n    # Currently the only validation we perform is that GBK operations have\n    # their coders set properly.\n    if transform_proto.spec.urn == common_urns.primitives.GROUP_BY_KEY.urn:\n      if len(transform_proto.inputs) != 1:\n        raise ValueError(\"Unexpected number of inputs: %s\" % transform_proto)\n      if len(transform_proto.outputs) != 1:\n        raise ValueError(\"Unexpected number of outputs: %s\" % transform_proto)\n      input_coder = get_coder(next(iter(transform_proto.inputs.values())))\n      output_coder = get_coder(next(iter(transform_proto.outputs.values())))\n      if input_coder.spec.urn != common_urns.coders.KV.urn:\n        raise ValueError(\n            \"Bad coder for input of %s: %s\" % (transform_id, input_coder))\n      if output_coder.spec.urn != common_urns.coders.KV.urn:\n        raise ValueError(\n            \"Bad coder for output of %s: %s\" % (transform_id, output_coder))\n      input_key_coder_id = input_coder.component_coder_ids[0]\n      output_key_coder_id = output_coder.component_coder_ids[0]\n      if input_key_coder_id != output_key_coder_id:\n        raise ValueError(\n            \"Input key coder %s does not match output key coder %s for \"\n            \"transform %s\" %\n            (input_key_coder_id, output_key_coder_id, transform_id))\n      output_values_coder_id = output_coder.component_coder_ids[1]\n      output_values_coder = pipeline_proto.components.coders[\n          output_values_coder_id]","sourceCodeStart":69,"sourceCodeEnd":105,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/pipeline_utils.py#L69-L105","documentation":"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.","triggerScenarios":"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.","commonSituations":"Custom runner or graph-rewrite tooling that fans out GBK output; serialized pipelines edited by external tools; portability graph conversion bugs between SDKs.","solutions":["Fix the failing transform so it has exactly one entry in transform_proto.outputs","Rebuild the pipeline from Beam code rather than constructing/altering protos manually","If fanning out results, insert explicit ParDo/flatten transforms after GBK instead of multiple GBK outputs","Align SDK/runner versions and re-serialize the pipeline"],"exampleFix":"// before\ngbk.spec.outputs['main'] = pc1\ngbk.spec.outputs['extra'] = pc2  # 2 outputs -> raises\n// after\ndel gbk.spec.outputs['extra']\nassert len(gbk.spec.outputs) == 1","handlingStrategy":"validation","validationCode":"def check_gbk_outputs(t: beam_runner_api_pb2.PTransform) -> bool:\n    return (t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn\n            or len(t.outputs) == 1)","typeGuard":"def is_wellformed_gbk(t) -> bool:\n    return t.spec.urn != 'beam:transform:group_by_key:v1' or len(t.outputs) == 1","tryCatchPattern":"try:\n    validate_pipeline_graph(pipeline_proto)\nexcept ValueError as e:\n    if 'Unexpected number of outputs' in str(e):\n        rewrite_gbk_to_single_output(pipeline_proto)\n    else:\n        raise","preventionTips":["Model fan-out with explicit downstream transforms, not multi-output GBK","Validate graph protos with validate_pipeline_graph before submission","Keep graph-rewrite tools conforming to the Beam model spec","Avoid hand-editing pipeline protos"],"tags":["python","apache-beam","pipeline-validation","group-by-key"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}