apache/beam · error · ValueError

Input key coder does not match output key coder for…

Error message

Input key coder %s does not match output key coder %s for transform %s

What it means

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.

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

Example fix

// before
out_kv = make_kv_coder(other_key_coder_id, values_coder_id)
// after
out_kv = make_kv_coder(input_coder.component_coder_ids[0], values_coder_id)

Defensive patterns

Strategy: validation

Validate before calling

def check_key_coders_match(pipeline, t) -> bool:
    if t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn:
        return True
    in_c = get_coder(next(iter(t.inputs.values())))
    out_c = get_coder(next(iter(t.outputs.values())))
    return in_c.component_coder_ids[0] == out_c.component_coder_ids[0]

Type guard

def keys_consistent(pipeline, t) -> bool:
    in_c, out_c = gbk_coders(pipeline, t)
    return in_c.component_coder_ids[0] == out_c.component_coder_ids[0]

Try / catch

try:
    validate_pipeline_graph(pipeline_proto)
except ValueError as e:
    if 'does not match output key coder' in str(e):
        unify_key_coder(pipeline_proto)
    else:
        raise

Prevention

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

When it happens

Trigger: 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.

Common situations: Coder rewriting during graph optimization; manually assigned coder IDs in tooling; pipelines merged from multiple proto fragments with clashing coder IDs.

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

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/ea0fd0db10df4fd9. Report an issue: GitHub.

Appendix: source

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

    # 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]
      if output_values_coder.spec.urn != common_urns.coders.ITERABLE.urn:
        raise ValueError(
            "Output value coder %s for transform %s must be an iterable "
            "coder, but uses URN %s" % (
                output_values_coder_id,
                transform_id,
                output_values_coder.spec.urn))
      input_value_coder_id = input_coder.component_coder_ids[1]
      output_value_coder_id = output_values_coder.component_coder_ids[0]
      if output_value_coder_id != input_value_coder_id:
        raise ValueError(
            "Input value coder %s does not match output value coder %s for "

View on GitHub (pinned to 12126d8942)