apache/beam · error · TransformError

PCollection not part of a pipeline.

Error message

PCollection not part of a pipeline.

What it means

_check_pcollection also verifies the PCollection belongs to a pipeline. A PCollection-like object whose .pipeline is None/falsy triggers TransformError, because transforms must run within a pipeline context to schedule steps.

Solutions

1. Produce PCollections via pipeline | transform instead of instantiating them directly
2. Use the output of the same active pipeline the transform will run in
3. Recreate the PCollection under the current pipeline if the old pipeline was discarded
4. In tests, build with TestPipeline and real transform application

Example fix

// before
pcoll = pvalue.PCollection(None)  # no pipeline attached
my_transform.expand(pcoll)
// after
pcoll = pipeline | beam.Create([...])
my_transform.expand(pcoll)

if pcoll is None or pcoll.pipeline is None:
    pcoll = active_pipeline | 'Recreate' >> beam.Create(source_items)

def is_attached_pcollection(x):
    return isinstance(x, pvalue.PCollection) and x.pipeline is not None

try:
    my_transform.expand(pcoll)
except error.TransformError as ex:
    logger.error('PCollection not attached to a pipeline: %s', ex)
    raise

When it happens

Trigger: Constructing or deserializing a PCollection manually without attaching it to an applied pipeline; using a PCollection from a different/pipeline-scoped context after the pipeline was discarded; pickling/dill issues that dropped the pipeline reference.

Common situations: Creating PCollection objects directly in unit tests without p | transform; cross-pipeline reuse of outputs; corrupted notebook state where the pipeline object was rebuilt but old PCollections retained.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Execution of [ ] not implemented in runner .
• Expected a PTransform object, got
• Expecting a PCollection argument.
• A BigQuery table or a query must be specified
• A cluster_identifier should be Optional[Union[str…

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/ptransform.py:571

  def __repr__(self):
    return '<%s at %s>' % (self._str_internal(), hex(id(self)))

  def _str_internal(self):
    return '%s(PTransform)%s%s%s' % (
        self.__class__.__name__,
        ' label=[%s]' % self.label if
        (hasattr(self, 'label') and self.label) else '',
        ' inputs=%s' % str(self.inputs) if
        (hasattr(self, 'inputs') and self.inputs) else '',
        ' side_inputs=%s' % str(self.side_inputs) if self.side_inputs else '')

  def _check_pcollection(self, pcoll):
    # type: (pvalue.PCollection) -> None
    if not isinstance(pcoll, pvalue.PCollection):
      raise error.TransformError('Expecting a PCollection argument.')
    if not pcoll.pipeline:
      raise error.TransformError('PCollection not part of a pipeline.')

  def get_windowing(self, inputs):
    # type: (Any) -> Windowing

    """Returns the window function to be associated with transform's output.

    By default most transforms just return the windowing function associated
    with the input PCollection (or the first input if several).
    """
    if inputs:
      return inputs[0].windowing
    else:
      from apache_beam.transforms.core import Windowing
      from apache_beam.transforms.window import GlobalWindows

      # TODO(robertwb): Return something compatible with every windowing?
      return Windowing(GlobalWindows())

View on GitHub (pinned to 12126d8942)