{"record":{"id":"37c20ada72368c94","repo":"apache/beam","slug":"bad-coder-for-output-of-s-s","errorCode":null,"errorMessage":"Bad coder for output of %s: %s","messagePattern":"Bad coder for output of (.+?): (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/pipeline_utils.py","lineNumber":94,"sourceCode":"        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]\n      if output_values_coder.spec.urn != common_urns.coders.ITERABLE.urn:\n        raise ValueError(\n            \"Output value coder %s for transform %s must be an iterable \"\n            \"coder, but uses URN %s\" % (\n                output_values_coder_id,\n                transform_id,\n                output_values_coder.spec.urn))","sourceCodeStart":76,"sourceCodeEnd":112,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/pipeline_utils.py#L76-L112","documentation":"Raised by validate_transform when the output coder of a GROUP_BY_KEY transform is not a KV coder. GBK must produce a KV<K, Iterable<V>> stream, so its output PCollection's coder must carry the KV URN. It indicates broken coder declarations in the pipeline proto's coder registry.","triggerScenarios":"A GBK transform whose single output PCollection resolves to a coder whose spec.urn is not common_urns.coders.KV.urn — typically after custom coder rewrites or hand-edited pipeline protos.","commonSituations":"Graph-optimization tools replacing post-GBK coders; custom runner output coder assignment; cross-language pipelines where the output coder URN was translated incorrectly.","solutions":["Set the GBK output coder to a KV coder whose components are (key coder, iterable-of-values coder)","Restore the coder originally inferred by Beam (rebuild the pipeline without the custom coder rewrite)","Verify custom translator/runner code assigns the correct output coder ID to the GBK output PCollection","Upgrade/re-serialize with the current SDK"],"exampleFix":"// before\npipeline.components.coders[gbk_out_coder_id].spec.urn = CUSTOM_URN\n// after\npipeline.components.coders[gbk_out_coder_id].spec.urn = \\\n    'beam:coders:kv:v1'  # KV(key, iterable(values))","handlingStrategy":"validation","validationCode":"def check_output_is_kv(pipeline, transform_id: str) -> bool:\n    t = pipeline.components.transforms[transform_id]\n    if t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn:\n        return True\n    out_pc = next(iter(t.outputs.values()))\n    coder = get_coder_for_pc(pipeline, out_pc)\n    return coder.spec.urn == common_urns.coders.KV.urn","typeGuard":"def output_is_kv(pipeline, t) -> bool:\n    out_pc = next(iter(t.outputs.values()))\n    coder = pipeline.components.coders[find_coder_for_pc(pipeline, out_pc)]\n    return coder.spec.urn == common_urns.coders.KV.urn","tryCatchPattern":"try:\n    validate_pipeline_graph(pipeline_proto)\nexcept ValueError as e:\n    if 'Bad coder for output' in str(e):\n        restore_default_gbk_output_coder(pipeline_proto)\n    else:\n        raise","preventionTips":["Let Beam infer GBK output coders (KV[K, Iterable[V]]); don't override them","Snapshot/compare coder URNs before and after any graph-rewrite step","Test custom runners with validate_pipeline_graph on representative pipelines","Keep coder registries intact when merging pipeline fragments"],"tags":["python","apache-beam","pipeline-validation","coders"],"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"}