apache/beam · error · TypeError

Unable to identify type

Error message

Unable to identify type: {typ} specified on column: {col_name}. Please provide a valid type from the following: {_default_type_to_tensor_type_map.keys()}

What it means

TFTProcessHandler._get_raw_data_feature_spec_per_column builds a tf.io feature spec for each column and only recognizes a fixed set of types (numpy scalar subtypes and keys of _default_type_to_tensor_type_map). When a column's declared type is outside that set, it cannot be mapped to a TensorInfo and a TypeError is raised. This often happens when a type is wrapped in a container like list[int] without unpacking, which the code explicitly rejects earlier.

Solutions

  1. Change the column's type annotation to a supported type: int, float, bool, bytes, str, or the corresponding np.generic types.
  2. Unwrap container annotations: use the inner scalar type (int instead of list[int]) or pass the list types explicitly as supported.
  3. Check _default_type_to_tensor_type_map in handlers.py for the exact supported set before annotating the schema.
  4. If a custom type is required, pre-convert the column to a supported dtype in a prior beam.Map before MLTransform.

Example fix

// before
beam.Row(x=[1, 2, 3]).with_output_types(...)  # column typed as list[int] container

// after
beam.Row(x=1).with_output_types(...)  # bare supported scalar type, e.g. int / np.int64
Defensive patterns

Strategy: type-guard

Validate before calling

import numpy as np
from apache_beam.ml.transforms.handlers import _default_type_to_tensor_type_map

def is_supported_column_type(typ) -> bool:
    return (isinstance(typ, type) and issubclass(typ, np.generic)) or typ in _default_type_to_tensor_type_map

assert all(is_supported_column_type(t) for t in my_column_types)

Type guard

def is_supported_column_type(typ) -> bool:
    import numpy as np
    from apache_beam.ml.transforms.handlers import _default_type_to_tensor_type_map
    return (isinstance(typ, type) and issubclass(typ, np.generic)) or typ in _default_type_to_tensor_type_map

Try / catch

try:
    result = pcoll | MLTransform(...).with_write_artifact_location(loc)
except TypeError as e:
    if 'Unable to identify type' in str(e):
        # fix schema annotation or convert column to a supported dtype
        raise ValueError(f'Unsupported column type in schema: {e}') from e
    raise

Prevention

When it happens

Trigger: Passing an MLTransform (TFT-based) a PCollection schema column whose type annotation is not a np.generic subclass nor a key of _default_type_to_tensor_type_map — e.g. a custom Python class, list[int] passed as the whole annotation where a bare type is expected, or str vs np.bytes_ mismatches.

Common situations: Declaring beam.Row schemas with Python typing wrappers (Optional[int], list[int]), using custom types in typed DoBags, or typos in type names when constructing TransformConfigs.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/ml/transforms/handlers.py:271

    primitive_containers_type = (
        list,
        collections.abc.Sequence,
    )
    is_primitive_container = (
        typing.get_origin(typ) in primitive_containers_type)

    if is_primitive_container:
      dtype = typing.get_args(typ)[0]
      if len(typing.get_args(typ)) > 1 or typing.get_origin(dtype) == Union:
        raise RuntimeError(
            f"Union type is not supported for column: {col_name}. "
            f"Please pass a PCollection with valid schema for column "
            f"{col_name} by passing a single type "
            "in container. For example, list[int].")
    elif issubclass(typ, np.generic) or typ in _default_type_to_tensor_type_map:
      dtype = typ
    else:
      raise TypeError(
          f"Unable to identify type: {typ} specified on column: {col_name}. "
          f"Please provide a valid type from the following: "
          f"{_default_type_to_tensor_type_map.keys()}")
    return tf.io.VarLenFeature(_default_type_to_tensor_type_map[dtype])

  def get_raw_data_metadata(
      self, input_types: dict[str, type]) -> dataset_metadata.DatasetMetadata:
    raw_data_feature_spec = self.get_raw_data_feature_spec(input_types)
    raw_data_feature_spec[_TEMP_KEY] = tf.io.VarLenFeature(dtype=tf.string)
    return self.convert_raw_data_feature_spec_to_dataset_metadata(
        raw_data_feature_spec)

  def write_transform_artifacts(self, transform_fn, location):
    """
    Write transform artifacts to the given location.
    Args:
      transform_fn: A transform_fn object.
      location: A location to write the artifacts.

View on GitHub (pinned to 12126d8942)