apache/beam · error · RuntimeError

Transforms must be instances of MLTransformProvider and…

Error message

Transforms must be instances of MLTransformProvider and implement get_ptransform_for_processing() method.

What it means

create_ptransform_list validates that every transform passed to MLTransform is an instance of MLTransformProvider and can produce a PTransform via get_ptransform_for_processing(). If any element in the transforms list is not such a provider (e.g. a raw PTransform or arbitrary callable), RuntimeError is raised before any processing happens.

Solutions

Only pass instances whose class subclasses MLTransformProvider and implements get_ptransform_for_processing().
Wrap your custom logic: class MyTransform(MLTransformProvider): def get_ptransform_for_processing(self, **kwargs): return MyPTransform().
Move non-ML Beam transforms outside MLTransform in the pipeline (before/after the MLTransform step).

Example fix

// before
MLTransform().with_transform(beam.Map(lambda x: x))
// after
class MyTransform(MLTransformProvider):
  def get_ptransform_for_processing(self, **kwargs):
    return beam.Map(lambda x: x)
MLTransform().with_transform(MyTransform())

Defensive patterns

Strategy: type-guard

Validate before calling

from apache_beam.ml.transforms.base import MLTransformProvider
bad = [t for t in transforms if not isinstance(t, MLTransformProvider)]
assert not bad, f'Not MLTransformProvider: {bad}'

Type guard

def is_ml_transform(t) -> bool:
    return isinstance(t, MLTransformProvider) and hasattr(t, 'get_ptransform_for_processing')

Try / catch

try:
    create_ptransform_list()
except RuntimeError as e:
    if 'MLTransformProvider' in str(e):
        transforms = [wrap(t) for t in transforms]
    else:
        raise

Prevention

Only pass MLTransformProvider instances to MLTransform.
Subclass MLTransformProvider for custom transforms and implement get_ptransform_for_processing().
Keep plain Beam PTransforms outside the MLTransform transforms list.
Unit-test custom transform classes instantiate and return a PTransform.

When it happens

Trigger: Passing a plain apache_beam.transforms.PTransform, a lambda, or a non-ML transform class instance into MLTransform(transforms=[...]) instead of instances of MLTransformProvider subclasses (e.g. built-in embeddings/handlers).

Common situations: Mixing generic Beam transforms (Map, ParDo) into the MLTransform transforms list; writing a custom transform but forgetting to subclass MLTransformProvider; passing a class where an instance is required or vice versa.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/ml/transforms/base.py:633

      pipeline_options: Optional[PipelineOptions] = None,
  ):
    self.transforms = transforms
    self._parent_artifact_location = artifact_location
    self.artifact_mode = artifact_mode
    self.pipeline_options = pipeline_options

  def create_and_save_ptransform_list(self):
    ptransform_list = self.create_ptransform_list()
    self.save_transforms_in_artifact_location(ptransform_list)
    return ptransform_list

  def create_ptransform_list(self):
    previous_ptransform_type = None
    current_ptransform = None
    ptransform_list = []
    for transform in self.transforms:
      if not isinstance(transform, MLTransformProvider):
        raise RuntimeError(
            'Transforms must be instances of MLTransformProvider and '
            'implement get_ptransform_for_processing() method.')
      # for each instance of PTransform, create a new artifact location
      current_ptransform = transform.get_ptransform_for_processing(
          artifact_location=os.path.join(
              self._parent_artifact_location, uuid.uuid4().hex[:6]),
          artifact_mode=self.artifact_mode)
      append_transform = hasattr(current_ptransform, 'append_transform')
      if (type(current_ptransform)
          != previous_ptransform_type) or not append_transform:
        ptransform_list.append(current_ptransform)
        previous_ptransform_type = type(current_ptransform)
      # If different PTransform is appended to the list and the PTransform
      # supports append_transform, append the transform to the PTransform.
      if append_transform:
        ptransform_list[-1].append_transform(transform)
    return ptransform_list

View on GitHub (pinned to 12126d8942)