{"record":{"id":"e0098e8eb76e62c0","repo":"apache/beam","slug":"input-value-to-a-stateful-dofn-or-keyparam-must-be-a-kv","errorCode":null,"errorMessage":"Input value to a stateful DoFn or KeyParam must be a KV tuple; instead, got '%s'.","messagePattern":"Input value to a stateful DoFn or KeyParam must be a KV tuple; instead, got '(.+?)'\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/common.py","lineNumber":1015,"sourceCode":"        window = GlobalWindow()\n      side_inputs = [si[window] for si in self.side_inputs]\n      side_inputs.extend(additional_args)\n      args_for_process, kwargs_for_process = util.insert_values_in_args(\n          self.args_for_process, self.kwargs_for_process, side_inputs)\n      if not self.recalculate_window_args:\n        self.args_for_process, self.kwargs_for_process = (\n            args_for_process, kwargs_for_process)\n        self.has_cached_window_args = True\n\n    # Extract key in the case of a stateful DoFn. Note that in the case of a\n    # stateful DoFn, we set during __init__ self.has_windowed_inputs to be\n    # True. Therefore, windows will be exploded coming into this method, and\n    # we can rely on the window variable being set above.\n    if self.user_state_context or self.is_key_param_required:\n      try:\n        key, unused_value = windowed_value.value\n      except (TypeError, ValueError):\n        raise ValueError((\n            'Input value to a stateful DoFn or KeyParam must be a KV tuple; '\n            'instead, got \\'%s\\'.') % (windowed_value.value, ))\n\n    for i, p in self.placeholders_for_process:\n      if core.DoFn.ElementParam == p:\n        args_for_process[i] = windowed_value.value\n      elif core.DoFn.KeyParam == p:\n        args_for_process[i] = key\n      elif core.DoFn.WindowParam == p:\n        args_for_process[i] = window\n      elif core.DoFn.WindowedValueParam == p:\n        args_for_process[i] = windowed_value\n      elif core.DoFn.TimestampParam == p:\n        args_for_process[i] = windowed_value.timestamp\n      elif core.DoFn.PaneInfoParam == p:\n        args_for_process[i] = windowed_value.pane_info\n      elif isinstance(p, core.DoFn.StateParam):\n        assert self.user_state_context is not None","sourceCodeStart":997,"sourceCodeEnd":1033,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/common.py#L997-L1033","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Key the input PCollection first (beam.WithKeys(...) or beam.Map(lambda x: (x.key, x)))","Confirm the element is exactly a 2-tuple with a hashable key","If statefulness is not needed, remove StateParam/TimerParam/KeyParam usage from the DoFn"],"exampleFix":"# before\nbeam.ParDo(MyStatefulDoFn())\n# after\nbeam.ParDo(MyStatefulDoFn()).with_input_types = None  # key first:\n(pcoll | beam.WithKeys(lambda x: x['id']) | beam.ParDo(MyStatefulDoFn()))","handlingStrategy":"validation","validationCode":"def ensure_keyed(elements):\n    bad = [e for e in elements if not (isinstance(e, tuple) and len(e) == 2)]\n    if bad:\n        raise ValueError('Stateful DoFn input must be (key, value) tuples; got %r' % bad[:3])\n\n# or in the pipeline:\npcoll = pcoll | beam.WithKeys(lambda x: x['id'])","typeGuard":"def is_kv(value):\n    return isinstance(value, tuple) and len(value) == 2 and isinstance(value[0], collections.abc.Hashable)","tryCatchPattern":"try:\n    result = pcoll | beam.ParDo(StatefulDoFn())\nexcept ValueError as e:\n    if 'must be a KV tuple' in str(e):\n        logging.error('Input not keyed; add beam.WithKeys upstream')\n    raise","preventionTips":["Always add beam.WithKeys before stateful transforms","Add a beam.Map asserting KV shape in tests","Keep element schemas documented in pipeline code"],"tags":["apache-beam","python","kv-tuple","stateful-dofn"],"backgroundTag":"invalid-argument-format","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}