apache/beam · error · ValueError

Coder for the GroupByKey operation "%s" is not a key-value c

Error message

Coder for the GroupByKey operation "%s" is not a key-value coder: %s.

What it means

Before submitting a GroupByKey, the DataflowRunner verifies that the coder for its input PCollection is a key-value (KV) coder and that the key coder is deterministic. A non-KV coder means the pipeline's type hints imply elements that aren't (key, value) pairs, so the shuffle cannot be encoded — a ValueError naming the GBK transform and the coder is raised.

Source

Thrown at sdks/python/apache_beam/runners/dataflow/dataflow_runner.py:554

    """Returns a coder based on a typehint object."""
    if window_coder:
      return coders.WindowedValueCoder(
          coders.registry.get_coder(typehint), window_coder=window_coder)
    return coders.registry.get_coder(typehint)

  def _verify_gbk_coders(self, transform, pcoll):
    # Infer coder of parent.
    #
    # TODO(ccy): make Coder inference and checking less specialized and more
    # comprehensive.

    parent = pcoll.producer
    if parent:
      coder = parent.transform._infer_output_coder()  # pylint: disable=protected-access
    if not coder:
      coder = self._get_coder(pcoll.element_type or typehints.Any, None)
    if not coder.is_kv_coder():
      raise ValueError((
          'Coder for the GroupByKey operation "%s" is not a '
          'key-value coder: %s.') % (transform.label, coder))
    # TODO(robertwb): Update the coder itself if it changed.
    coders.registry.verify_deterministic(
        coder.key_coder(), 'GroupByKey operation "%s"' % transform.label)

  def get_default_gcp_region(self):
    """Get a default value for Google Cloud region according to
    https://cloud.google.com/compute/docs/gcloud-compute/#default-properties.
    If no default can be found, returns None.
    """
    environment_region = os.environ.get('CLOUDSDK_COMPUTE_REGION')
    if environment_region:
      _LOGGER.info(
          'Using default GCP region %s from $CLOUDSDK_COMPUTE_REGION',
          environment_region)
      return environment_region
    try:

View on GitHub (pinned to 12126d8942)

Solutions

  1. Map elements to KV pairs before the GBK: pairs = beam.Map(lambda x: (x['k'], x)).
  2. Add explicit type hints, e.g. beam.Map(fn).with_output_types(types.KV[str, int]).
  3. Check the producing transform's output coder with pipeline apply/inspect to confirm it's KV.
  4. If keys use a non-deterministic coder, register a deterministic key coder or use a hashable/deterministic key representation.

Example fix

// before
result = words | beam.GroupByKey()
// after
result = (words
    | beam.Map(lambda w: (w, 1)).with_output_types(typing.Tuple[str, int])
    | beam.GroupByKey())
Defensive patterns

Strategy: validation

Validate before calling

from apache_beam import typehints
from apache_beam.coders import typecoders
coder = typecoders.registry.get_coder(pcoll.element_type or typehints.Any)
assert coder.is_kv_coder(), 'PCollection fed to GroupByKey must be KV[A, B]'

Try / catch

try:
    result = pcoll | beam.GroupByKey()
except ValueError as e:
    if 'not a key-value coder' in str(e):
        result = (pcoll | beam.Map(lambda x: (x['k'], x))) | beam.GroupByKey()

Prevention

When it happens

Trigger: Running beam.GroupByKey() on a PCollection whose inferred/declared element type isn't KV[A,B] — e.g. applying GBK directly on a PCollection of scalars, or producing pairs via a transform whose output coder wasn't inferred as KV.

Common situations: Forgetting to Map to (key, value) pairs before GroupByKey; using lambdas/custom DoFns that erase type information; map side outputs with Any type hints.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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