apache/beam · error · ValueError

No logical type registered for typing '%s'

Error message

No logical type registered for typing '%s'

What it means

LogicalType.from_typing() looks up a registered LogicalType implementation that can encode the given Python typing. If none is registered it raises ValueError. Beam only converts typings to schema logical types for which a LogicalType subclass has been registered.

Source

Thrown at sdks/python/apache_beam/typehints/schemas.py:900

  def register_logical_type(cls, logical_type_cls):
    """Register an implementation of LogicalType."""
    cls._known_logical_types.add(logical_type_cls.urn(), logical_type_cls)
    return logical_type_cls

  @classmethod
  def from_typing(cls, typ):
    # type: (type) -> LogicalType

    """Construct an instance of a registered LogicalType implementation given a
    typing.

    Raises ValueError if no registered LogicalType implementation can encode the
    given typing."""

    logical_type = cls._known_logical_types.get_logical_type_by_language_type(
        typ)
    if logical_type is None:
      raise ValueError("No logical type registered for typing '%s'" % typ)

    return logical_type._from_typing(typ)

  @classmethod
  def _from_typing(cls, typ):
    # type: (type) -> LogicalType

    """Construct an instance of this LogicalType implementation given a typing.
    """
    raise NotImplementedError()

  @classmethod
  def from_runner_api(cls, logical_type_proto):
    # type: (schema_pb2.LogicalType) -> LogicalType

    """Construct an instance of a registered LogicalType implementation given a
    proto LogicalType.

View on GitHub (pinned to 12126d8942)

Solutions

1. Import the module that registers the logical type (e.g. apache_beam.typehints.schemas) before conversion.
2. Implement and register a LogicalType subclass via LogicalType.register_logical_type(MyLogicalType) supporting the typing.
3. Fall back to a plain supported type (str/bytes/int) in the schema instead of the custom typing.

Example fix

// before
lt = LogicalType.from_typing(MyCustomType)  # ValueError

// after
LogicalType.register_logical_type(MyCustomLogicalType)  # registers language_type=MyCustomType
lt = LogicalType.from_typing(MyCustomType)

from apache_beam.typehints.schemas import LogicalType

def has_logical_type(typ):
    return LogicalType._known_logical_types.get_logical_type_by_language_type(typ) is not None

try:
    lt = LogicalType.from_typing(typ)
except ValueError:
    lt = None  # use a supported plain type instead
if lt is None:
    typ = bytes  # fallback encoding

When it happens

Trigger: Calling LogicalType.from_typing(SomeType) with a typing (e.g. a custom class, datetime with custom params) that no registered LogicalType's language_type() matches; using a logical type in a schema before its register_logical_type() call ran.

Common situations: Using a custom logical type whose registration module was never imported (side-effect registration missing); Beam version differences where a logical type (e.g. ParameterizedType, thift/decimal variants) is not registered by default.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• reconciling for ToType
• Unable to convert logical type ${identifier}
• Schema with id {schema.id} has encoding_positions_set=True,
• Attempted to encode null for non-nullable field "{}".
• Encountered a type that is not currently supported by RowCod

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