apache/beam · error · ValueError

Bad coder for input of

Error message

Bad coder for input of %s: %s

What it means

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.

Solutions

1. 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
2. Apply beam KV coders explicitly via pvalue.with_output_types or set the coder spec URN to KV in the proto
3. Check that custom coder wiring/translation code preserves KV coders on GBK inputs
4. Re-serialize the pipeline with a current Beam SDK

Example fix

// before
pc | beam.GroupByKey()  # input is not KV
// after
pc = items | beam.Map(lambda x: (x['key'], x))  | beam.GroupByKey()

def check_input_is_kv(pipeline, transform_id: str) -> bool:
    t = pipeline.components.transforms[transform_id]
    if t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn:
        return True
    coder = pipeline.components.coders[next(iter(t.inputs.values())) and get_coder_id(pipeline, t)]
    return coder.spec.urn == common_urns.coders.KV.urn

def input_is_kv(pipeline, t) -> bool:
    in_pc = next(iter(t.inputs.values()))
    coder = pipeline.components.coders[find_coder_for_pc(pipeline, in_pc)]
    return coder.spec.urn == common_urns.coders.KV.urn

try:
    validate_pipeline_graph(pipeline_proto)
except ValueError as e:
    if 'Bad coder for input' in str(e):
        fix_gbk_input_to_kv(pipeline_proto)
    else:
        raise

When it happens

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

Common situations: 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.

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

• Bad coder for output of
• Input key coder does not match output key coder for…
• Input value coder does not match output value coder for…
• Output value coder for transform must be an iterable coder…
• Encountered a type that is not currently supported by…

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

Appendix: source

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

  """
  def get_coder(pcoll_id):
    return pipeline_proto.components.coders[
        pipeline_proto.components.pcollections[pcoll_id].coder_id]

  def validate_transform(transform_id):
    transform_proto = pipeline_proto.components.transforms[transform_id]

    # 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" % (

View on GitHub (pinned to 12126d8942)