What causes "Decode not implemented: %s." in apache/beam?

Implement decode(self, encoded) in your Coder subclass

apache/beam · error · NotImplementedError

Decode not implemented: %s.

Error message

Decode not implemented: %s.

What it means

Coder.decode is abstract in apache_beam; the base class raises NotImplementedError('Decode not implemented: %s.') when a Coder subclass does not override decode. Decoding happens whenever the pipeline materializes elements from encoded bytes (checkpoint marks, side inputs, retries), so this surfaces at read time.

Source

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

def deserialize_coder(serialized):
  from apache_beam.internal import pickler
  return pickler.loads(serialized.split(b'$', 1)[1], use_zlib=True)


# pylint: enable=wrong-import-order, wrong-import-position


class Coder(object):
  """Base class for coders."""
  def encode(self, value):
    # type: (Any) -> bytes

    """Encodes the given object into a byte string."""
    raise NotImplementedError('Encode not implemented: %s.' % self)

  def decode(self, encoded):
    """Decodes the given byte string into the corresponding object."""
    raise NotImplementedError('Decode not implemented: %s.' % self)

  def encode_nested(self, value):
    """Uses the underlying implementation to encode in nested format."""
    return self.get_impl().encode_nested(value)

  def decode_nested(self, encoded):
    """Uses the underlying implementation to decode in nested format."""
    return self.get_impl().decode_nested(encoded)

  def is_deterministic(self):
    # type: () -> bool

    """Whether this coder is guaranteed to encode values deterministically.

    A deterministic coder is required for key coders in GroupByKey operations
    to produce consistent results.

    For example, note that the default coder, the PickleCoder, is not

View on GitHub (pinned to 12126d8942)

Solutions

Implement decode(self, encoded) in your Coder subclass
Use ToBytesCoder only for write paths and a real coder where decoding is needed
Round-trip test your coder (beam.WindowValue/beam.Coder test helpers) before running pipelines

Example fix

// before
class MyCoder(Coder):
    def encode(self, value):
        return str(value).encode()
// after
class MyCoder(Coder):
    def encode(self, value):
        return str(value).encode()
    def decode(self, encoded):
        return int(encoded.decode())

Defensive patterns

Strategy: type-guard

Validate before calling

if type(coder).decode is Coder.decode:
    raise TypeError('coder does not implement decode')

Type guard

def is_decodable(coder):
    return callable(getattr(coder, 'decode', None)) and type(coder).decode is not Coder.decode

Try / catch

try:
    value = coder.decode(blob)
except NotImplementedError as e:
    log.error('coder %s cannot decode: %s', coder, e)
    value = None  # or use an alternative reader

Prevention

Never use write-only coders on read paths
Unit-test encode->decode round trips
Subclass concrete coders rather than the abstract Coder

When it happens

Trigger: Calling coder.decode(bytes) on a base Coder or subclass that omitted decode; triggered by checkpoint readers, decodeFromBase64, bag page values, or pipeline code that round-trips encoded values.

Common situations: Custom coders implemented with only encode; coders for write-only usage later reused for reading; accidental use of base Coder instead of a concrete one.

Related errors

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