apache/beam · error · ValueError

Input elements to the transform

Error message

Input elements to the transform %s with stateful DoFn must be key-value pairs.

What it means

Raised by a stateful transform's expand() in apache_beam.transforms/util.py when its input PCollection's coder is not a key-value coder. Stateful DoFns (state/timers) require input elements to be (key, value) tuples so each state cell is scoped by key; the pipeline cannot determine keys from a non-KV coder, so Beam fails fast at graph-construction time.

Solutions

Map input to (key, value) tuples before the stateful transform: beam.Map(lambda x: (x['id'], x)).
Add explicit type hints to the preceding PTransform, e.g. @with_input_types(k=v_type) or beam.Map(fn).with_output_types(typehints.KV[k_type, v_type]), so Beam derives a KV coder.
For custom key classes, implement a deterministic Coder for the key so is_kv_coder() and key_coder() succeed.
Inspect the input coder (pCollection.element_type / coders.registry) to confirm the element type is KV before applying the stateful transform.

Example fix

# before
result = pcoll | 'stateful' | beam.ParDo(MyStatefulDoFn())  # pcoll yields dicts

# after
result = (pcoll
    | beam.Map(lambda x: (x['user_id'], x)).with_output_types(
        typehints.KV[str, dict])
    | 'stateful' | beam.ParDo(MyStatefulDoFn()))

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam import typehints
from apache_beam import coders
if pc.element_type is not None:
    assert typehints.is_consistent_with(pc.element_type, typehints.KV[typehints.Any, typehints.Any]), 'stateful DoFn input must be KV'

Type guard

def is_kv_typed(element_type):
    return element_type is not None and typehints.is_consistent_with(
        element_type, typehints.KV[typehints.Any, typehints.Any])

Try / catch

try:
    out = pcoll | stateful_transform
except ValueError as e:
    if 'must be key-value pairs' in str(e):
        pcoll = pcoll | beam.Map(lambda x: (x['key'], x))
        out = pcoll | stateful_transform
    else:
        raise

Prevention

Always emit explicit (key, value) tuples before any stateful DoFn.
Annotate with with_output_types(typehints.KV[K, V]) so coder inference cannot fail silently.
Provide deterministic Coders for custom key classes.
Add a pipeline-construction unit test that builds the stateful stage so coder errors surface in CI, not at runtime.

When it happens

Trigger: Applying a stateful DoFn via a wrapper (e.g. with_stateful_do_fn / BatchElements-like stateful transforms) to a PCollection of non-pair elements, or to KV pairs whose coder Beam cannot infer as a KV (missing/ambiguous type hints preceding a GroupByKey step in the same expand path).

Common situations: Passing plain dicts, lists, or single values into a stateful transform; producing KV pairs from a map without type hints so Beam's coder inference yields a non-KV coder; chaining a stateful transform after a source whose element type is only discovered at runtime.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/util.py:480

  def expand(self, pcoll):
    key_type, value_type = (typehints.typehints.coerce_to_kv_type(
        pcoll.element_type).tuple_types)
    kv_type_hint = typehints.KV[key_type, value_type]
    if kv_type_hint and kv_type_hint != typehints.Any:
      coder = coders.registry.get_coder(kv_type_hint)
      try:
        coder = coder.as_deterministic_coder(self.label)
      except ValueError:
        _LOGGER.warning(
            'GroupByEncryptedKey %s: '
            'The key coder is not deterministic. This may result in incorrect '
            'pipeline output. This can be fixed by adding a type hint to the '
            'operation preceding the GroupByKey step, and for custom key '
            'classes, by writing a deterministic custom Coder. Please see the '
            'documentation for more details.',
            self.label)
      if not coder.is_kv_coder():
        raise ValueError(
            'Input elements to the transform %s with stateful DoFn must be '
            'key-value pairs.' % self)
      key_coder = coder.key_coder()
      value_coder = coder.value_coder()
    else:
      key_coder = coders.registry.get_coder(typehints.Any)
      value_coder = key_coder

    gbk = beam.GroupByKey()
    gbk._inside_gbek = True
    output_type = tuple[key_type, Iterable[value_type]]

    return (
        pcoll
        | beam.ParDo(_EncryptMessage(self._hmac_key, key_coder, value_coder))
        | gbk
        | beam.ParDo(
            _DecryptMessage(self._hmac_key, key_coder,

View on GitHub (pinned to 12126d8942)