apache/beam · error · ValueError

Output value coder for transform must be an iterable coder…

Error message

Output value coder %s for transform %s must be an iterable coder, but uses URN %s

What it means

Raised by validate_transform: the value component of a GROUP_BY_KEY output KV coder must be an ITERABLE coder (GBK yields KV<K, Iterable<V>>), but the proto references a coder with a different URN. This keeps the post-GBK encoding contract intact for runners that rely on iterable streaming semantics.

Solutions

1. Wrap the value component in an iterable coder (beam:coders:iterable:v1) whose element coder is the input value coder
2. Rebuild the pipeline in Beam Python and let TypeHints infer KV[K, Iterable[V]] coders
3. Fix custom graph-rewrite code so GBK output coders are always KV(key_coder, iterable(value_coder))
4. Re-serialize with a current SDK

Example fix

// before
out_kv = make_kv_coder(key_coder_id, value_coder_id)  # plain value coder
// after
iter_coder = make_iterable_coder(value_coder_id)
out_kv = make_kv_coder(key_coder_id, iter_coder)

def check_value_coder_is_iterable(pipeline, t) -> bool:
    if t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn:
        return True
    out_c = get_coder(next(iter(t.outputs.values())))
    values_coder = pipeline.components.coders[out_c.component_coder_ids[1]]
    return values_coder.spec.urn == common_urns.coders.ITERABLE.urn

def gbk_values_iterable(pipeline, t) -> bool:
    out_c = gbk_output_coder(pipeline, t)
    return pipeline.components.coders[out_c.component_coder_ids[1]].spec.urn \
        == 'beam:coders:iterable:v1'

try:
    validate_pipeline_graph(pipeline_proto)
except ValueError as e:
    if 'must be an iterable coder' in str(e):
        wrap_in_iterable_coder(pipeline_proto)
    else:
        raise

When it happens

Trigger: output_values_coder.spec.urn != common_urns.coders.ITERABLE.urn for the GBK output KV's second component — e.g. the output coder was built with a plain list/sequence coder or an unrelated coder ID.

Common situations: Custom coder construction when rewriting GBK; hand-built protos; cross-language graph conversion assigning the wrong value coder component.

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 input of
• 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…
• Encountered a type that is not currently supported by…

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

Appendix: source

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

      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 "
            "transform %s" %
            (input_value_coder_id, output_value_coder_id, transform_id))
    elif transform_proto.spec.urn == common_urns.primitives.ASSIGN_WINDOWS.urn:
      if not transform_proto.inputs:
        raise ValueError("Missing input for transform: %s" % transform_proto)
    elif transform_proto.spec.urn == common_urns.primitives.PAR_DO.urn:
      if not transform_proto.inputs:
        raise ValueError("Missing input for transform: %s" % transform_proto)

View on GitHub (pinned to 12126d8942)