What causes "Bad coder for output of %s: %s" in apache/beam?

Set the GBK output coder to a KV coder whose components are (key coder, iterable-of-values coder)

apache/beam · error · ValueError

Bad coder for output of

Error message

Bad coder for output of %s: %s

What it means

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.

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

Example fix

// before
pipeline.components.coders[gbk_out_coder_id].spec.urn = CUSTOM_URN
// after
pipeline.components.coders[gbk_out_coder_id].spec.urn = \
    'beam:coders:kv:v1'  # KV(key, iterable(values))

Defensive patterns

Strategy: validation

Validate before calling

def check_output_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
    out_pc = next(iter(t.outputs.values()))
    coder = get_coder_for_pc(pipeline, out_pc)
    return coder.spec.urn == common_urns.coders.KV.urn

Type guard

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

Try / catch

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

Prevention

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

When it happens

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

Common situations: Graph-optimization tools replacing post-GBK coders; custom runner output coder assignment; cross-language pipelines where the output coder URN was translated incorrectly.

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/37c20ada72368c94. Report an issue: GitHub.

Appendix: source

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

        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" % (
                output_values_coder_id,
                transform_id,
                output_values_coder.spec.urn))

View on GitHub (pinned to 12126d8942)