{"record":{"id":"5d74c1da588ad162","repo":"apache/beam","slug":"input-elements-to-the-transform-s-with-stateful-dofn-must-be-5d74c1","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/util.py","lineNumber":480,"sourceCode":"  def expand(self, pcoll):\n    key_type, value_type = (typehints.typehints.coerce_to_kv_type(\n        pcoll.element_type).tuple_types)\n    kv_type_hint = typehints.KV[key_type, value_type]\n    if kv_type_hint and kv_type_hint != typehints.Any:\n      coder = coders.registry.get_coder(kv_type_hint)\n      try:\n        coder = coder.as_deterministic_coder(self.label)\n      except ValueError:\n        _LOGGER.warning(\n            'GroupByEncryptedKey %s: '\n            'The key coder is not deterministic. This may result in incorrect '\n            'pipeline output. This can be fixed by adding a type hint to the '\n            'operation preceding the GroupByKey step, and for custom key '\n            'classes, by writing a deterministic custom Coder. Please see the '\n            'documentation for more details.',\n            self.label)\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      value_coder = coder.value_coder()\n    else:\n      key_coder = coders.registry.get_coder(typehints.Any)\n      value_coder = key_coder\n\n    gbk = beam.GroupByKey()\n    gbk._inside_gbek = True\n    output_type = tuple[key_type, Iterable[value_type]]\n\n    return (\n        pcoll\n        | beam.ParDo(_EncryptMessage(self._hmac_key, key_coder, value_coder))\n        | gbk\n        | beam.ParDo(\n            _DecryptMessage(self._hmac_key, key_coder,","sourceCodeStart":462,"sourceCodeEnd":498,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/util.py#L462-L498","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","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."],"exampleFix":"# before\nresult = pcoll | 'stateful' | beam.ParDo(MyStatefulDoFn())  # pcoll yields dicts\n\n# after\nresult = (pcoll\n    | beam.Map(lambda x: (x['user_id'], x)).with_output_types(\n        typehints.KV[str, dict])\n    | 'stateful' | beam.ParDo(MyStatefulDoFn()))","handlingStrategy":"validation","validationCode":"from apache_beam import typehints\nfrom apache_beam import coders\nif pc.element_type is not None:\n    assert typehints.is_consistent_with(pc.element_type, typehints.KV[typehints.Any, typehints.Any]), 'stateful DoFn input must be KV'","typeGuard":"def is_kv_typed(element_type):\n    return element_type is not None and typehints.is_consistent_with(\n        element_type, typehints.KV[typehints.Any, typehints.Any])","tryCatchPattern":"try:\n    out = pcoll | stateful_transform\nexcept ValueError as e:\n    if 'must be key-value pairs' in str(e):\n        pcoll = pcoll | beam.Map(lambda x: (x['key'], x))\n        out = pcoll | stateful_transform\n    else:\n        raise","preventionTips":["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."],"tags":["python","apache-beam","stateful-dofn","type-hints","kv-coder"],"backgroundTag":"type-mismatch","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"}