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

Implement encode(self, value) returning bytes in your Coder subclass

apache/beam · error · NotImplementedError

Encode not implemented: %s.

Error message

Encode not implemented: %s.

What it means

Coder.encode is abstract in apache_beam; the base class raises NotImplementedError('Encode not implemented: %s.') whenever a subclass fails to override encode. This means an instance of a coder class that never implemented encoding was used where the pipeline needed to serialize an element.

Source

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

      coder.__class__.__name__.encode('utf-8'),
      pickler.dumps(coder, use_zlib=True))


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.

View on GitHub (pinned to 12126d8942)

Solutions

Implement encode(self, value) returning bytes in your Coder subclass
Use a built-in coder (coders.BytesCoder, ToBytesCoder, etc.) instead of a bare Coder
Check that you instantiate the concrete coder class, not the abstract base

Example fix

// before
class MyCoder(Coder):
    def decode(self, encoded):
        return json.loads(encoded)
// after
class MyCoder(Coder):
    def encode(self, value):
        return json.dumps(value).encode('utf-8')
    def decode(self, encoded):
        return json.loads(encoded)

Defensive patterns

Strategy: type-guard

Validate before calling

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

Type guard

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

Try / catch

try:
    data = coder.encode(value)
except NotImplementedError as e:
    log.error('coder %s cannot encode: %s', coder, e)
    data = fallback_coder.encode(value)

Prevention

Always implement both encode and decode in custom coders
Round-trip test custom coders in unit tests
Use built-in coders when possible

When it happens

Trigger: Calling coder.encode(value) on the base Coder class or a custom subclass that only overrides decode/other methods; passing a custom coder object to DoFn/transform APIs that then tries to serialize elements.

Common situations: Users writing custom coders forgetting to override encode (often only decode was implemented); using an abstract/base Coder directly in tests or pipelines; refactors that renamed a coder's encode method.

Related errors

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