{"record":{"id":"053ccfcb348cc668","repo":"apache/beam","slug":"bad-coder-for-input-of-s-s","errorCode":null,"errorMessage":"Bad coder for input of %s: %s","messagePattern":"Bad coder for input of (.+?): (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/pipeline_utils.py","lineNumber":91,"sourceCode":"  \"\"\"\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]\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\" % (","sourceCodeStart":73,"sourceCodeEnd":109,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/pipeline_utils.py#L73-L109","documentation":"Raised by validate_transform when the input coder of a GROUP_BY_KEY transform is not a KV coder. GBK operates on KV-encoded pcollections, so its input PCollection must use a coder with the KV URN. This surfaces malformed coder wiring in the pipeline proto, usually from custom code or external graph manipulation.","triggerScenarios":"A GBK transform whose single input PCollection resolves to a coder whose spec.urn is not common_urns.coders.KV.urn — e.g. the input was created from a plain dict/tuple without an explicit KV coder, or coder IDs were rewritten by tooling.","commonSituations":"Manually building pipeline protos; custom coders applied to GBK inputs; portability pipelines where coders were re-assigned during graph conversion; older pipelines serialized with nonstandard coders.","solutions":["Ensure the GBK input PCollection is a KV<K,V> (e.g. beam.Map(lambda x: (x['k'], x)) before GroupByKey) so the coder infers as KV","Apply beam KV coders explicitly via pvalue.with_output_types or set the coder spec URN to KV in the proto","Check that custom coder wiring/translation code preserves KV coders on GBK inputs","Re-serialize the pipeline with a current Beam SDK"],"exampleFix":"// before\npc | beam.GroupByKey()  # input is not KV\n// after\npc = items | beam.Map(lambda x: (x['key'], x))  | beam.GroupByKey()","handlingStrategy":"validation","validationCode":"def check_input_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    coder = pipeline.components.coders[next(iter(t.inputs.values())) and get_coder_id(pipeline, t)]\n    return coder.spec.urn == common_urns.coders.KV.urn","typeGuard":"def input_is_kv(pipeline, t) -> bool:\n    in_pc = next(iter(t.inputs.values()))\n    coder = pipeline.components.coders[find_coder_for_pc(pipeline, in_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 input' in str(e):\n        fix_gbk_input_to_kv(pipeline_proto)\n    else:\n        raise","preventionTips":["Ensure elements are KV pairs before GroupByKey (apply beam.Map to produce tuples)","Add explicit type hints (pc.with_output_types(KV[K, V])) so coders infer correctly","Check pc.element_type before GBK in pipelines built programmatically","Avoid custom coders on GBK inputs unless they are proper KV coders"],"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"}