apache/beam · error · ValueError

All PCollections must belong to the same pipeline.

Error message

All PCollections must belong to the same pipeline.

What it means

ib.collect() materializes PCollections through a single recording/compute session scoped to one pipeline. Before computing, it groups all input PCollections by pcoll.pipeline; if they resolve to more than one distinct pipeline it raises this ValueError. Interactive Beam cannot drive two different pipelines in one collect call.

Source

Thrown at sdks/python/apache_beam/runners/interactive/interactive_beam.py:938

    return ()

  def as_pcollection(pcoll_or_df):
    if isinstance(pcoll_or_df, DeferredBase):
      # Get the proxy so we can get the output shape of the DataFrame.
      pcoll, element_type = deferred_df_to_pcollection(pcoll_or_df)
      watch({'anonymous_pcollection_{}'.format(id(pcoll)): pcoll})
      return pcoll, element_type
    elif isinstance(pcoll_or_df, beam.pvalue.PCollection):
      return pcoll_or_df, pcoll_or_df.element_type
    else:
      raise TypeError(f'{pcoll} is not an apache_beam.pvalue.PCollection.')

  pcolls_with_element_types = [as_pcollection(p) for p in pcolls]
  pcolls_to_element_types = dict(pcolls_with_element_types)
  pcolls = [pcoll for pcoll, _ in pcolls_with_element_types]
  pipelines = set(pcoll.pipeline for pcoll in pcolls)
  if len(pipelines) != 1:
    raise ValueError('All PCollections must belong to the same pipeline.')
  pipeline, = pipelines

  if isinstance(n, str):
    assert n == 'inf', (
        'Currently only the string \'inf\' is supported. This denotes reading '
        'elements until the recording is stopped via a kernel interrupt.')
  elif isinstance(n, int):
    assert n > 0, 'n needs to be positive or the string \'inf\''

  if isinstance(duration, int):
    assert duration > 0, ('duration needs to be positive, a duration string, '
                          'or the string \'inf\'')

  if n == 'inf':
    n = float('inf')

  if duration == 'inf':
    duration = float('inf')

View on GitHub (pinned to 12126d8942)

Solutions

1. Create all PCollections from the same pipeline object; reuse one pipeline variable across cells.
2. Move every transform onto a single pipeline: pc2 = pipeline | 'Step2' >> ... instead of pipeline2 | ....
3. If you accidentally re-created the pipeline, re-run the earlier cells so all PCollections bind to the newest pipeline object.
4. Call ib.collect once per pipeline, or merge pipelines if they are logically one job.
5. Ensure you didn't pass a PCollection from a pipeline built with a different runner.

Example fix

// before: p1 = beam.Pipeline(); p2 = beam.Pipeline(); ib.collect(a_from_p1, b_from_p2) | // after: p = beam.Pipeline(); a = p | 'A' >> beam.Create([1]); b = p | 'B' >> beam.Create([2]); ib.collect(a, b)

def same_pipeline(pcolls): return len({p.pipeline for p in pcolls}) == 1; assert same_pipeline(pcolls)

def belongs_to(pcoll, pipeline): return pcoll.pipeline is pipeline

try: df = ib.collect(*pcolls) | except ValueError as e: (print({id(p.pipeline) for p in pcolls}) if 'same pipeline' in str(e) else None); raise

When it happens

Trigger: ib.collect(pcoll_a, pcoll_b) where pcoll_a was created on pipeline1 and pcoll_b on pipeline2 (two beam.Pipeline() instances, or one from a re-executed cell); building PCollections in two notebook cells each calling beam.Pipeline() then collecting across them.

Common situations: Notebooks: re-running a cell that does pipeline = beam.Pipeline() creates a NEW pipeline each run, so PCollections from an old run live on a different pipeline object than fresh ones; mixing DirectRunner pipelines with the interactive pipeline; merging outputs of two experiments into one collect.

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

• The given pcoll %s is not a dict, an iterable or a PCollecti
• The given pcoll {pcoll_container} is not a dict, an iterable
• The beam_sql magic tries to query PCollections from multiple
• cache_root GCS bucket path is invalid.
• PCollection not available, please run the pipeline.

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