apache/beam · error · TypeCheckError

Input to _GroupByKeyOnly must be a PCollection of windowed k

Error message

Input to _GroupByKeyOnly must be a PCollection of windowed key-value pairs. Instead received: %r.

What it means

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.

Source

Thrown at sdks/python/apache_beam/runners/direct/transform_evaluator.py:998

        self._applied_ptransform.outputs[None].element_type or
        self._applied_ptransform.transform.get_type_hints().input_types[0][0])
    self.key_coder = coders.registry.get_coder(kv_type_hint.tuple_types[0])

  def process_timer(self, timer_firing):
    # We do not need to emit a KeyedWorkItem to process_element().
    pass

  def process_element(self, element):
    assert not self.global_state.get_state(
        None, _GroupByKeyOnlyEvaluator.COMPLETION_TAG)
    if (isinstance(element, WindowedValue) and
        isinstance(element.value, abc.Iterable) and len(element.value) == 2):
      k, v = element.value
      encoded_k = self.key_coder.encode(k)
      state = self._step_context.get_keyed_state(encoded_k)
      state.add_state(None, _GroupByKeyOnlyEvaluator.ELEMENTS_TAG, v)
    else:
      raise TypeCheckError(
          'Input to _GroupByKeyOnly must be a PCollection of '
          'windowed key-value pairs. Instead received: %r.' % element)

  def finish_bundle(self):
    if self._is_final_bundle():
      if self.global_state.get_state(None,
                                     _GroupByKeyOnlyEvaluator.COMPLETION_TAG):
        # Ignore empty bundles after emitting output. (This may happen because
        # empty bundles do not affect input watermarks.)
        bundles = []
      else:
        gbk_result = []
        # TODO(ccy): perhaps we can clean this up to not use this
        # internal attribute of the DirectStepContext.
        for encoded_k in self._step_context.existing_keyed_state:
          # Ignore global state.
          if encoded_k is None:
            continue

View on GitHub (pinned to 12126d8942)

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.

Example fix

// before
result = (pcoll | beam.GroupByKey())
// after
result = (pcoll | beam.Map(lambda x: (x['user_id'], x)) | beam.GroupByKey())

Defensive patterns

Strategy: type-guard

Validate before calling

def is_kv(el):
    from apache_beam.transforms.window import WindowedValue
    return isinstance(el, WindowedValue) and isinstance(el.value, tuple) and len(el.value) == 2
# assert all(is_kv(e) for e in sample_elements) before applying GroupByKey

Type guard

def is_windowed_kv(el):
    from apache_beam.transforms.window import WindowedValue
    return isinstance(el, WindowedValue) and isinstance(el.value, abc.Iterable) and len(el.value) == 2

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/6d14da1a4e87d7d5. Report an issue: GitHub.