apache/beam · error · NotImplementedError

key_coder: %s

Error message

key_coder: %s

What it means

Coder.key_coder() raises NotImplementedError when is_kv_coder() is True but the coder has no key-component notion implemented (base-class default), i.e. the coder claims to be a KV coder yet does not provide a key coder. It raises ValueError('Not a KV coder') otherwise.

Source

Thrown at sdks/python/apache_beam/coders/coders.py:261

    # TODO: After https://github.com/apache/beam/issues/18490 we should be
    # able to infer the type hint rather than require every subclass define
    # it
    raise NotImplementedError

  @classmethod
  def from_type_hint(cls, unused_typehint, unused_registry):
    # type: (Type[CoderT], Any, CoderRegistry) -> CoderT
    # If not overridden, just construct the coder without arguments.
    return cls()

  def is_kv_coder(self):
    # type: () -> bool
    return False

  def key_coder(self):
    # type: () -> Coder
    if self.is_kv_coder():
      raise NotImplementedError('key_coder: %s' % self)
    else:
      raise ValueError('Not a KV coder: %s.' % self)

  def value_coder(self):
    # type: () -> Coder
    if self.is_kv_coder():
      raise NotImplementedError('value_coder: %s' % self)
    else:
      raise ValueError('Not a KV coder: %s.' % self)

  def _get_component_coders(self):
    # type: () -> Sequence[Coder]

    """For internal use only; no backwards-compatibility guarantees.

    Returns the internal component coders of this coder."""
    # This is an internal detail of the Coder API and does not need to be
    # refined in user-defined Coders.

View on GitHub (pinned to 12126d8942)

Solutions

1. Override key_coder()/value_coder() or _get_component_coders() in your custom coder
2. Use a concrete KV coder (e.g. coders.TupleCoder) that supplies component coders
3. Ensure is_kv_coder() accurately reflects your coder (return False if it is not a KV coder)

Example fix

// before
class MyCoder(Coder):
    def is_kv_coder(self):
        return True
// after
class MyCoder(Coder):
    def is_kv_coder(self):
        return False
    # or implement:
    def key_coder(self):
        return coders.BytesCoder()

if coder.is_kv_coder() and type(coder).key_coder is Coder.key_coder:
    raise TypeError('coder claims KV but lacks key_coder')

def has_key_coder(coder):
    return coder.is_kv_coder() and type(coder).key_coder is not Coder.key_coder

try:
    kc = coder.key_coder()
except NotImplementedError as e:
    log.error('no key coder for %s: %s', coder, e)
    kc = default_key_coder

When it happens

Trigger: Calling key_coder() on a custom Coder whose is_kv_coder() returns True (default True in some paths) but which does not override _get_component_coders/key_coder; base Coder instances used where the framework asks for the key coder (e.g. for grouping).

Common situations: Custom coders for KV-shaped values that never implemented the component coder API; runner code or user code probing coders for key/value split on non-KV-aware coders.

Related errors

• Not a KV coder: %s.
• value_coder: %s
• cannot encode a null KV
• ToBytesCoder cannot be used for decoding.
• Typehint is not of nullable type, and cannot be converted to

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