apache/beam · error · ValueError

Input value to a stateful DoFn or KeyParam must be a KV…

Error message

Input value to a stateful DoFn or KeyParam must be a KV tuple; instead, got '%s'.

What it means

Stateful DoFns and DoFns using KeyParam operate per key, so Beam requires each input element to be a (key, value) tuple. When unpacking windowed_value.value into key/value fails with TypeError or ValueError, Beam raises this ValueError telling the developer the element shape is wrong.

Solutions

1. Key the input PCollection first (beam.WithKeys(...) or beam.Map(lambda x: (x.key, x)))
2. Confirm the element is exactly a 2-tuple with a hashable key
3. If statefulness is not needed, remove StateParam/TimerParam/KeyParam usage from the DoFn

Example fix

# before
beam.ParDo(MyStatefulDoFn())
# after
beam.ParDo(MyStatefulDoFn()).with_input_types = None  # key first:
(pcoll | beam.WithKeys(lambda x: x['id']) | beam.ParDo(MyStatefulDoFn()))

def ensure_keyed(elements):
    bad = [e for e in elements if not (isinstance(e, tuple) and len(e) == 2)]
    if bad:
        raise ValueError('Stateful DoFn input must be (key, value) tuples; got %r' % bad[:3])

# or in the pipeline:
pcoll = pcoll | beam.WithKeys(lambda x: x['id'])

def is_kv(value):
    return isinstance(value, tuple) and len(value) == 2 and isinstance(value[0], collections.abc.Hashable)

try:
    result = pcoll | beam.ParDo(StatefulDoFn())
except ValueError as e:
    if 'must be a KV tuple' in str(e):
        logging.error('Input not keyed; add beam.WithKeys upstream')
    raise

When it happens

Trigger: Applying a stateful DoFn (or one with a KeyParam process argument) to a PCollection whose elements are not 2-tuples, e.g. plain scalars, dicts, or 3-tuples.

Common situations: Forgetting a beam.KeyValue mapping/WithKeys step before a stateful transform; upstream output changed shape after a refactor; running with --streaming or user-state features enabled on a pipeline whose elements were never keyed.

Understand the failure class

Background: "Invalid ... format", "must be in format X", "does not look like a ..." — invalid argument format errors across CLI tools and libraries — this error's family across 17 libraries.

Related errors

• DoFn %r has multiple StateSpecs with the same name
• DoFn has unsupported per-key DoFn param . Per-key DoFn…
• Input elements to the transform
• Input elements to the transform
• A BigQuery table or a query must be specified

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/common.py:1015

        window = GlobalWindow()
      side_inputs = [si[window] for si in self.side_inputs]
      side_inputs.extend(additional_args)
      args_for_process, kwargs_for_process = util.insert_values_in_args(
          self.args_for_process, self.kwargs_for_process, side_inputs)
      if not self.recalculate_window_args:
        self.args_for_process, self.kwargs_for_process = (
            args_for_process, kwargs_for_process)
        self.has_cached_window_args = True

    # Extract key in the case of a stateful DoFn. Note that in the case of a
    # stateful DoFn, we set during __init__ self.has_windowed_inputs to be
    # True. Therefore, windows will be exploded coming into this method, and
    # we can rely on the window variable being set above.
    if self.user_state_context or self.is_key_param_required:
      try:
        key, unused_value = windowed_value.value
      except (TypeError, ValueError):
        raise ValueError((
            'Input value to a stateful DoFn or KeyParam must be a KV tuple; '
            'instead, got \'%s\'.') % (windowed_value.value, ))

    for i, p in self.placeholders_for_process:
      if core.DoFn.ElementParam == p:
        args_for_process[i] = windowed_value.value
      elif core.DoFn.KeyParam == p:
        args_for_process[i] = key
      elif core.DoFn.WindowParam == p:
        args_for_process[i] = window
      elif core.DoFn.WindowedValueParam == p:
        args_for_process[i] = windowed_value
      elif core.DoFn.TimestampParam == p:
        args_for_process[i] = windowed_value.timestamp
      elif core.DoFn.PaneInfoParam == p:
        args_for_process[i] = windowed_value.pane_info
      elif isinstance(p, core.DoFn.StateParam):
        assert self.user_state_context is not None

View on GitHub (pinned to 12126d8942)