apache/beam · error · TypeError

Expected a PTransform object, got

Error message

Expected a PTransform object, got %s

What it means

Pipeline.apply/_apply_internal requires the transform argument to be an instance of PTransform. Anything else (a function, class, string, None) cannot be applied to the pipeline, so a TypeError is raised naming the offending object.

Solutions

1. Instantiate the transform: apply(Map(fn)) not apply(Map).
2. Wrap plain functions in beam.Map/beam.FlatMap/beam.ParDo instead of passing them directly.
3. Check argument order: pipeline.apply(transform, pvalueish, label).
4. Verify the variable isn't None due to a failed factory/import.

Example fix

// before
result = pipeline.apply(beam.Map)
// after
result = pipeline.apply(beam.Map(lambda x: x * 2))

from apache_beam.transforms.ptransform import PTransform
assert isinstance(my_transform, PTransform), 'not a PTransform instance'

def is_ptransform(t) -> bool:
    from apache_beam.transforms.ptransform import PTransform
    return isinstance(t, PTransform)

try:
    pipeline.apply(t, pcoll)
except TypeError as e:
    if 'Expected a PTransform' in str(e):
        raise ValueError(f'wrap {t!r} in beam.Map/beam.ParDo') from e

When it happens

Trigger: pipeline.apply(MyTransformClass) (class not instance), pipeline.apply(some_function), pipeline.apply('label', pcoll) with arguments in the wrong order, or passing the result of a factory that returned None.

Common situations: Forgetting parentheses when instantiating a transform; passing a plain function instead of wrapping it (e.g. beam.Map(fn)); mixing up apply's positional signature apply(transform, pvalueish, label).

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• A cluster_identifier should be Optional[Union[str…
• Cannot get a type descriptor for
• Cannot interpret as Duration.
• CombineGlobally can be used only with combineFn objects…
• database_config must be VectorDatabaseWriteConfig, got

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

Appendix: source

Thrown at sdks/python/apache_beam/pipeline.py:725

    with scoped_pipeline_options(self._options):
      return self._apply_internal(transform, pvalueish, label)

  def _apply_internal(
      self,
      transform: ptransform.PTransform,
      pvalueish: Optional[pvalue.PValue] = None,
      label: Optional[str] = None) -> pvalue.PValue:
    """Internal implementation of apply(), called within scoped options."""
    if isinstance(transform, ptransform._NamedPTransform):
      return self.apply(
          transform.transform, pvalueish, label or transform.label)

    if not label and isinstance(transform, ptransform._PTransformFnPTransform):
      # This must be set before label is inspected.
      transform.set_options(self._options)

    if not isinstance(transform, ptransform.PTransform):
      raise TypeError("Expected a PTransform object, got %s" % transform)

    if label:
      # Fix self.label as it is inspected by some PTransform operations
      # (e.g. to produce error messages for type hint violations).
      old_label, transform.label = transform.label, label
      try:
        return self.apply(transform, pvalueish)
      finally:
        transform.label = old_label

    # Attempts to alter the label of the transform to be applied only when it's
    # a top-level transform so that the cell number will not be prepended to
    # every child transform in a composite.
    if self._current_transform() is self._root_transform():
      alter_label_if_ipython(transform, pvalueish)

    full_label = '/'.join(
        [self._current_transform().full_label, transform.label]).lstrip('/')

View on GitHub (pinned to 12126d8942)