{"record":{"id":"ea0fd0db10df4fd9","repo":"apache/beam","slug":"input-key-coder-s-does-not-match-output-key-coder-s-for","errorCode":null,"errorMessage":"Input key coder %s does not match output key coder %s for transform %s","messagePattern":"Input key coder (.+?) does not match output key coder (.+?) for transform (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/pipeline_utils.py","lineNumber":99,"sourceCode":"    # 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))\n      input_value_coder_id = input_coder.component_coder_ids[1]\n      output_value_coder_id = output_values_coder.component_coder_ids[0]\n      if output_value_coder_id != input_value_coder_id:\n        raise ValueError(\n            \"Input value coder %s does not match output value coder %s for \"","sourceCodeStart":81,"sourceCodeEnd":117,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/pipeline_utils.py#L81-L117","documentation":"Raised by validate_transform when the key coder of a GROUP_BY_KEY's input KV differs from the key coder of its output KV. GBK must preserve key encodings across the transform; a mismatch means the pipeline graph would decode/encode keys inconsistently. It's a model-level consistency check over the coder component registry.","triggerScenarios":"input_coder.component_coder_ids[0] (input key coder ID) != output_coder.component_coder_ids[0] (output key coder ID) for a GBK transform — e.g. someone swapped or regenerated the output coder with a different key component.","commonSituations":"Coder rewriting during graph optimization; manually assigned coder IDs in tooling; pipelines merged from multiple proto fragments with clashing coder IDs.","solutions":["Make the GBK output KV coder reuse the input's key coder component ID","Rebuild the pipeline from Beam code so coders are inferred consistently end-to-end","Audit any custom code that modifies coder component IDs and keep the key coder stable across GBK","Re-serialize the pipeline with a single consistent SDK version"],"exampleFix":"// before\nout_kv = make_kv_coder(other_key_coder_id, values_coder_id)\n// after\nout_kv = make_kv_coder(input_coder.component_coder_ids[0], values_coder_id)","handlingStrategy":"validation","validationCode":"def check_key_coders_match(pipeline, t) -> bool:\n    if t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn:\n        return True\n    in_c = get_coder(next(iter(t.inputs.values())))\n    out_c = get_coder(next(iter(t.outputs.values())))\n    return in_c.component_coder_ids[0] == out_c.component_coder_ids[0]","typeGuard":"def keys_consistent(pipeline, t) -> bool:\n    in_c, out_c = gbk_coders(pipeline, t)\n    return in_c.component_coder_ids[0] == out_c.component_coder_ids[0]","tryCatchPattern":"try:\n    validate_pipeline_graph(pipeline_proto)\nexcept ValueError as e:\n    if 'does not match output key coder' in str(e):\n        unify_key_coder(pipeline_proto)\n    else:\n        raise","preventionTips":["Reuse the input key coder component ID when constructing GBK output coders","Never regenerate coders for GBK outputs from scratch in graph tooling","Diff coder component IDs across graph rewrites in tests","Use a single SDK version to produce and consume pipeline protos"],"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"}