{"record":{"id":"084ef90694c728a6","repo":"apache/beam","slug":"input-elements-to-the-transform-s-with-stateful-dofn-must-be","errorCode":null,"errorMessage":"Input elements to the transform %s with stateful DoFn must be key-value pairs.","messagePattern":"Input elements to the transform (.+?) with stateful DoFn must be key-value pairs\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/core.py","lineNumber":1782,"sourceCode":"\n  def _process_argspec_fn(self):\n    return self.fn._process_argspec_fn()\n\n  def display_data(self):\n    return {\n        'fn': DisplayDataItem(self.fn.__class__, label='Transform Function'),\n        'fn_dd': self.fn\n    }\n\n  def expand(self, pcoll):\n    # In the case of a stateful DoFn, warn if the key coder is not\n    # deterministic.\n    if self._signature.is_stateful_dofn():\n      kv_type_hint = pcoll.element_type\n      if kv_type_hint and kv_type_hint != typehints.Any:\n        coder = coders.registry.get_coder(kv_type_hint)\n        if not coder.is_kv_coder():\n          raise ValueError(\n              'Input elements to the transform %s with stateful DoFn must be '\n              'key-value pairs.' % self)\n        key_coder = coder.key_coder()\n      else:\n        key_coder = coders.registry.get_coder(typehints.Any)\n\n      if not key_coder.is_deterministic():\n        _LOGGER.warning(\n            'Key coder %s for transform %s with stateful DoFn may not '\n            'be deterministic. This may cause incorrect behavior for complex '\n            'key types. Consider adding an input type hint for this transform.',\n            key_coder,\n            self)\n\n    if self._signature.is_unbounded_per_element():\n      is_bounded = False\n    else:\n      is_bounded = pcoll.is_bounded","sourceCodeStart":1764,"sourceCodeEnd":1800,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/core.py#L1764-L1800","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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)."],"exampleFix":"// before\nbeam.ParDo(MyStatefulDoFn())  # applied to PCollection[int]\n// after\n| beam.Map(lambda x: (x % 10, x))  # produce KV pairs\n| beam.ParDo(MyStatefulDoFn())","handlingStrategy":"validation","validationCode":"from apache_beam import typehints\nfrom apache_beam import coders\nhint = pcoll.element_type\ncoder = coders.registry.get_coder(hint if hint else typehints.Any)\nassert coder.is_kv_coder(), 'stateful DoFn needs KV input, got %s' % hint","typeGuard":"def is_kv_input(pcoll):\n    hint = pcoll.element_type\n    if not hint or hint == typehints.Any:\n        return True\n    return coders.registry.get_coder(hint).is_kv_coder()","tryCatchPattern":"null","preventionTips":["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."],"tags":["python","apache-beam","stateful-dofn","type-hints"],"backgroundTag":"invalid-argument-value","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"}