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

Apache Beam's stateful DoFn support (per-key state and timers) requires input elements to be key-value pairs, because state is scoped per key. Before executing the transform, Beam derives a key coder from the input PCollection's element type; if the type hint resolves to a coder that is not a KV coder, it raises this ValueError.

Solutions

Ensure the input PCollection's element type is a key-value pair (KV[type_key, type_value]) before the stateful ParDo, e.g. add a beam.Map(lambda x: (key, x)) step.
Annotate the input PCollection with an explicit KV type hint via papply or beam.Map with with_output_types=typehints.KV[k, v].
If the type hint is wrong, fix the upstream transform's with_output_types so coder.is_kv_coder() returns True.
If elements genuinely need no key, redesign: either pick a constant key or replace stateful logic with a non-stateful DoFn (or group-by-key based approach).

Example fix

// before
beam.ParDo(MyStatefulDoFn())  # applied to PCollection[int]
// after
| beam.Map(lambda x: (x % 10, x))  # produce KV pairs
| beam.ParDo(MyStatefulDoFn())

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam import typehints
from apache_beam import coders
hint = pcoll.element_type
coder = coders.registry.get_coder(hint if hint else typehints.Any)
assert coder.is_kv_coder(), 'stateful DoFn needs KV input, got %s' % hint

Type guard

def is_kv_input(pcoll):
    hint = pcoll.element_type
    if not hint or hint == typehints.Any:
        return True
    return coders.registry.get_coder(hint).is_kv_coder()

Try / catch

null

Prevention

Always key elements (Map to KV) immediately before stateful transforms.
Add with_output_types=typehints.KV[k, v] on the producing Map so mismatches fail early.
Unit-test stateful pipeline fragments with known element types to catch coder errors at test time.

When it happens

Trigger: Applying a stateful DoFn via ParDo (e.g. with @stateful/@timers in the DoFn) to a PCollection whose element type hint is not a KV type, such as elements of plain type str/int or a non-tuple class, while element_type is known and not Any.

Common situations: Developers write a stateful DoFn but forget to produce keyed output (e.g. beam.Map(lambda x: ('key', x)) missing upstream); type hints inferred as non-KV after a Map/FlatMap step; pipelines where beam.Window or grouping removed the KV structure.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/core.py:1782

  def _process_argspec_fn(self):
    return self.fn._process_argspec_fn()

  def display_data(self):
    return {
        'fn': DisplayDataItem(self.fn.__class__, label='Transform Function'),
        'fn_dd': self.fn
    }

  def expand(self, pcoll):
    # In the case of a stateful DoFn, warn if the key coder is not
    # deterministic.
    if self._signature.is_stateful_dofn():
      kv_type_hint = pcoll.element_type
      if kv_type_hint and kv_type_hint != typehints.Any:
        coder = coders.registry.get_coder(kv_type_hint)
        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()
      else:
        key_coder = coders.registry.get_coder(typehints.Any)

      if not key_coder.is_deterministic():
        _LOGGER.warning(
            'Key coder %s for transform %s with stateful DoFn may not '
            'be deterministic. This may cause incorrect behavior for complex '
            'key types. Consider adding an input type hint for this transform.',
            key_coder,
            self)

    if self._signature.is_unbounded_per_element():
      is_bounded = False
    else:
      is_bounded = pcoll.is_bounded

View on GitHub (pinned to 12126d8942)