{"record":{"id":"6d14da1a4e87d7d5","repo":"apache/beam","slug":"input-to-groupbykeyonly-must-be-a-pcollection-of-windowed","errorCode":null,"errorMessage":"Input to _GroupByKeyOnly must be a PCollection of windowed key-value pairs. Instead received: %r.","messagePattern":"Input to _GroupByKeyOnly must be a PCollection of windowed key-value pairs\\. Instead received: %r\\.","errorType":"validation","errorClass":"TypeCheckError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/direct/transform_evaluator.py","lineNumber":998,"sourceCode":"        self._applied_ptransform.outputs[None].element_type or\n        self._applied_ptransform.transform.get_type_hints().input_types[0][0])\n    self.key_coder = coders.registry.get_coder(kv_type_hint.tuple_types[0])\n\n  def process_timer(self, timer_firing):\n    # We do not need to emit a KeyedWorkItem to process_element().\n    pass\n\n  def process_element(self, element):\n    assert not self.global_state.get_state(\n        None, _GroupByKeyOnlyEvaluator.COMPLETION_TAG)\n    if (isinstance(element, WindowedValue) and\n        isinstance(element.value, abc.Iterable) and len(element.value) == 2):\n      k, v = element.value\n      encoded_k = self.key_coder.encode(k)\n      state = self._step_context.get_keyed_state(encoded_k)\n      state.add_state(None, _GroupByKeyOnlyEvaluator.ELEMENTS_TAG, v)\n    else:\n      raise TypeCheckError(\n          'Input to _GroupByKeyOnly must be a PCollection of '\n          'windowed key-value pairs. Instead received: %r.' % element)\n\n  def finish_bundle(self):\n    if self._is_final_bundle():\n      if self.global_state.get_state(None,\n                                     _GroupByKeyOnlyEvaluator.COMPLETION_TAG):\n        # Ignore empty bundles after emitting output. (This may happen because\n        # empty bundles do not affect input watermarks.)\n        bundles = []\n      else:\n        gbk_result = []\n        # TODO(ccy): perhaps we can clean this up to not use this\n        # internal attribute of the DirectStepContext.\n        for encoded_k in self._step_context.existing_keyed_state:\n          # Ignore global state.\n          if encoded_k is None:\n            continue","sourceCodeStart":980,"sourceCodeEnd":1016,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/direct/transform_evaluator.py#L980-L1016","documentation":"The direct runner's _GroupByKeyOnly evaluator receives each element and expects a WindowedValue whose value is a 2-tuple (key, value). If the element is not a windowed value or not a 2-item iterable, it raises TypeCheckError because a GroupByKey cannot proceed on malformed input.","triggerScenarios":"Feeding a PCollection that was not produced as (key, value) pairs into a GroupByKey, applying GroupByKey to a flat (non-tuple) collection, or elements wrapped incorrectly (not WindowedValue) in custom DoFns/sources feeding the GBK step.","commonSituations":"Mistakenly calling beam.GroupByKey() on a collection of scalars or 3-tuples; producing key-value pairs via a map that forgets to emit tuples; custom sources emitting raw values bypassing windowing; Python 2/3 iterables that are not 2-length.","solutions":["Ensure the PCollection immediately upstream of GroupByKey contains exactly 2-tuples (key, value), e.g. pipe | beam.Map(lambda x: (x['k'], x)).","Verify each element is a WindowedValue; for custom sources/DoFns wrap values with windowed_value or use beam.WindowInto so the runner receives windowed values.","Check element count of the tuple: use beam.Tuple below/flatMap emitting pairs of length 2 only.","Inspect the upstream transform's output type with beam.Map(print) or a type-check pipeline option (beam type checking) to catch the mismatch before the runner."],"exampleFix":"// before\nresult = (pcoll | beam.GroupByKey())\n// after\nresult = (pcoll | beam.Map(lambda x: (x['user_id'], x)) | beam.GroupByKey())","handlingStrategy":"type-guard","validationCode":"def is_kv(el):\n    from apache_beam.transforms.window import WindowedValue\n    return isinstance(el, WindowedValue) and isinstance(el.value, tuple) and len(el.value) == 2\n# assert all(is_kv(e) for e in sample_elements) before applying GroupByKey","typeGuard":"def is_windowed_kv(el):\n    from apache_beam.transforms.window import WindowedValue\n    return isinstance(el, WindowedValue) and isinstance(el.value, abc.Iterable) and len(el.value) == 2","tryCatchPattern":null,"preventionTips":["Always map to (key, value) 2-tuples immediately before GroupByKey.","Enable Beam runtime type checking to catch shape errors at pipeline build.","In custom sources/DoFns, emit windowed values rather than raw elements."],"tags":["apache-beam","groupbykey","type-check","python"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}