apache/beam · error · TypeError

Unable to deterministically encode '%s' of type '%s', please

Error message

Unable to deterministically encode '%s' of type '%s', please provide a type hint for the input of '%s'

What it means

In encode_special_deterministic, when a value is not a proto/frozen-dataclass/namedtuple/enum/stateful object, the coder falls back to raising TypeError via _deterministic_encoding_error_msg. Without a known structured type, Beam cannot guarantee reproducible bytes for objects like arbitrary class instances (it would otherwise pickle them, which is not deterministic). The message asks for a type hint so Beam can pick a deterministic coder.

Source

Thrown at sdks/python/apache_beam/coders/coder_impl.py:518

            "Unable to deterministically encode non-frozen '%s' of type '%s' "
            "for the input of '%s'" %
            (value, type(value), self.requires_deterministic_step_label))
      init_fields = [field for field in dataclasses.fields(value) if field.init]
      try:
        if any(field.kw_only for field in init_fields):
          stream.write_byte(DATACLASS_KW_ONLY_TYPE)
          self.encode_type(type(value), stream)
          stream.write_var_int64(len(init_fields))
          for field in init_fields:
            stream.write(field.name.encode("utf-8"), True)
            self.encode_to_stream(getattr(value, field.name), stream, True)
        else:  # Not using kw_only, we can pass parameters by position.
          stream.write_byte(DATACLASS_TYPE)
          self.encode_type(type(value), stream)
          values = [getattr(value, field.name) for field in init_fields]
          self.iterable_coder_impl.encode_to_stream(values, stream, True)
      except Exception as e:
        raise TypeError(self._deterministic_encoding_error_msg(value)) from e
    elif isinstance(value, tuple) and hasattr(type(value), '_fields'):
      stream.write_byte(NAMED_TUPLE_TYPE)
      self.encode_type(type(value), stream)
      try:
        self.iterable_coder_impl.encode_to_stream(value, stream, True)
      except Exception as e:
        raise TypeError(self._deterministic_encoding_error_msg(value)) from e
    elif isinstance(value, enum.Enum):
      stream.write_byte(ENUM_TYPE)
      self.encode_type(type(value), stream)
      # Enum values can be of any type.
      try:
        self.encode_to_stream(value.value, stream, True)
      except Exception as e:
        raise TypeError(self._deterministic_encoding_error_msg(value)) from e
    elif (hasattr(value, "__getstate__") and
          # https://github.com/apache/beam/issues/33020
          type(value).__reduce__ == object.__reduce__):

View on GitHub (pinned to 12126d8942)

Solutions

  1. Provide a type hint (e.g. input/output type on the DoFn or PTransform) matching a deterministically encodable type (str, bytes, int, Tuple, NamedTuple, frozen dataclass).
  2. Replace dict/set values with sorted tuples or NamedTuples before using them as keys.
  3. Define __getstate__/__setstate__ (and keep default __reduce__) on the class so the nested-state deterministic path is used.
  4. Wrap the value into a frozen dataclass or NamedTuple key.

Example fix

// before
class ExtractKey(beam.DoFn):
    def process(self, element):
        yield (element['meta'], element['value'])  # dict key

// after
class MetaKey(typing.NamedTuple):
    a: str
    b: int

class ExtractKey(beam.DoFn):
    def process(self, element) -> Tuple[MetaKey, int]:
        yield (MetaKey(element['a'], element['b']), element['value'])
Defensive patterns

Strategy: validation

Validate before calling

def is_deterministically_encodable(value):
    import dataclasses, enum
    if isinstance(value, (str, bytes, int, float, bool)):
        return True
    if dataclasses.is_dataclass(value) and type(value).__dataclass_params__.frozen:
        return True
    if isinstance(value, tuple) and hasattr(type(value), '_fields'):
        return all(is_deterministically_encodable(v) for v in value)
    if isinstance(value, enum.Enum):
        return is_deterministically_encodable(value.value)
    return False

Type guard

def is_encodable_key(value) -> bool:
    return isinstance(value, (str, bytes, int, float, bool)) or (isinstance(value, tuple) and hasattr(type(value), '_fields'))

Try / catch

try:
    coder.encode(key)
except TypeError as e:
    raise ValueError(f"Key {key!r} is not deterministically encodable; use a NamedTuple/frozen dataclass") from e

Prevention

When it happens

Trigger: Encoding an arbitrary class instance (or set/dict or other unstructured value) as the input to a determinism-requiring step (e.g. GroupByKey keys) with no type hint registered; the final else branch in encode_special_deterministic raises via self._deterministic_encoding_error_msg(value).

Common situations: Passing custom objects or sets as GroupByKey/CoGroupByKey keys; omitting type hints on DoFn outputs so Beam falls back to Any/pickle; dict/set used as a key which has no deterministic iteration order.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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