{"record":{"id":"79b3d0fb586c7760","repo":"apache/beam","slug":"input-to-groupbykey-must-be-a-pcollection-with-elements","errorCode":null,"errorMessage":"Input to GroupByKey must be a PCollection with elements compatible with KV[A, B]","messagePattern":"Input to GroupByKey must be a PCollection with elements compatible with KV\\[A, B\\]","errorType":"validation","errorClass":"TypeCheckError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/fn_api_runner/trigger_manager.py","lineNumber":67,"sourceCode":"from apache_beam.utils import windowed_value\nfrom apache_beam.utils.timestamp import MIN_TIMESTAMP\nfrom apache_beam.utils.timestamp import Timestamp\nfrom apache_beam.utils.windowed_value import WindowedValue\n\n_LOGGER = logging.getLogger(__name__)\n_LOGGER.setLevel(logging.DEBUG)\n\nK = typing.TypeVar('K')\n\n\nclass _ReifyWindows(DoFn):\n  \"\"\"Receives KV pairs, and wraps the values into WindowedValues.\"\"\"\n  def process(\n      self, element, window=DoFn.WindowParam, timestamp=DoFn.TimestampParam):\n    try:\n      k, v = element\n    except TypeError:\n      raise TypeCheckError(\n          'Input to GroupByKey must be a PCollection with '\n          'elements compatible with KV[A, B]')\n\n    yield (k, windowed_value.WindowedValue(v, timestamp, [window]))\n\n\nclass _GroupBundlesByKey(DoFn):\n  def start_bundle(self):\n    self.keys = defaultdict(list)\n\n  def process(self, element):\n    key, windowed_value = element\n    self.keys[key].append(windowed_value)\n\n  def finish_bundle(self):\n    for k, vals in self.keys.items():\n      yield windowed_value.WindowedValue((k, vals),\n                                         MIN_TIMESTAMP, [GlobalWindow()])","sourceCodeStart":49,"sourceCodeEnd":85,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/fn_api_runner/trigger_manager.py#L49-L85","documentation":"The KV-wrapping DoFn preceding a GroupByKey unpacks each element into a key/value pair. If the element is not a 2-tuple (or dict-like pair) it raises TypeCheckError. This means the PCollection feeding GroupByKey does not contain KV[A, B]-compatible elements.","triggerScenarios":"Calling pipeline | GroupByKey() (or p.value_as_dict etc.) on a PCollection whose elements are scalars, lists of the wrong arity, or custom objects without tuple semantics — e.g. after a Map that emits single values instead of (key, value) pairs.","commonSituations":"Forgetting to emit (key, value) tuples upstream (e.g. mapping to just values), a Map returning 3-tuples or single ints, or Python 2/3 style flat_map outputs that changed element arity.","solutions":["Ensure the transform immediately upstream of GroupByKey emits (key, value) tuples: use beam.Map(lambda x: (x['k'], x)) or beam.transforms.ptransform with beam.KV","Check with an assertion: add beam.Map(lambda kv: assert isinstance(kv, tuple) and len(kv)==2) before the GroupByKey in testing","If elements are single values, group by an index/key first (e.g. enumerate into (key, value) pairs)","Add a runtime type hint (with input_types or beam.PTransform type check) to catch the mismatch at pipeline construction time"],"exampleFix":"# before\npc | beam.Map(lambda x: x['value']) | beam.GroupByKey()  # elements are not KV\n# after\npc | beam.Map(lambda x: (x['key'], x['value'])) | beam.GroupByKey()","handlingStrategy":"type-guard","validationCode":"def kv_ok(pcoll):\n    return pcoll.element_type is None or pcoll.element_type == KV[Any, Any]\n# or runtime check on sample elements\nisinstance(elem, tuple) and len(elem) == 2","typeGuard":"def is_kv(element):\n    return isinstance(element, tuple) and len(element) == 2","tryCatchPattern":"try:\n    _ = pcoll | beam.GroupByKey()\nexcept TypeCheckError as e:\n    log.error('Bad GroupByKey input: %s', e)  # upstream must emit (k, v) tuples\n    raise","preventionTips":["Always Map to (key, value) tuples immediately before GroupByKey","Enable type hints (with_input_validation) to catch element-shape bugs at graph construction","Add a debug/assertion Map that checks tuple arity in tests","Review upstream Map/FlatMap lambdas whenever changing element shapes"],"tags":["python","apache-beam","groupbykey","type-check"],"backgroundTag":"type-mismatch","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"}