{"record":{"id":"38746558ff30eee7","repo":"apache/beam","slug":"all-pcollections-must-belong-to-the-same-pipeline","errorCode":null,"errorMessage":"All PCollections must belong to the same pipeline.","messagePattern":"All PCollections must belong to the same pipeline\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/interactive_beam.py","lineNumber":938,"sourceCode":"    return ()\n\n  def as_pcollection(pcoll_or_df):\n    if isinstance(pcoll_or_df, DeferredBase):\n      # Get the proxy so we can get the output shape of the DataFrame.\n      pcoll, element_type = deferred_df_to_pcollection(pcoll_or_df)\n      watch({'anonymous_pcollection_{}'.format(id(pcoll)): pcoll})\n      return pcoll, element_type\n    elif isinstance(pcoll_or_df, beam.pvalue.PCollection):\n      return pcoll_or_df, pcoll_or_df.element_type\n    else:\n      raise TypeError(f'{pcoll} is not an apache_beam.pvalue.PCollection.')\n\n  pcolls_with_element_types = [as_pcollection(p) for p in pcolls]\n  pcolls_to_element_types = dict(pcolls_with_element_types)\n  pcolls = [pcoll for pcoll, _ in pcolls_with_element_types]\n  pipelines = set(pcoll.pipeline for pcoll in pcolls)\n  if len(pipelines) != 1:\n    raise ValueError('All PCollections must belong to the same pipeline.')\n  pipeline, = pipelines\n\n  if isinstance(n, str):\n    assert n == 'inf', (\n        'Currently only the string \\'inf\\' is supported. This denotes reading '\n        'elements until the recording is stopped via a kernel interrupt.')\n  elif isinstance(n, int):\n    assert n > 0, 'n needs to be positive or the string \\'inf\\''\n\n  if isinstance(duration, int):\n    assert duration > 0, ('duration needs to be positive, a duration string, '\n                          'or the string \\'inf\\'')\n\n  if n == 'inf':\n    n = float('inf')\n\n  if duration == 'inf':\n    duration = float('inf')","sourceCodeStart":920,"sourceCodeEnd":956,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/interactive_beam.py#L920-L956","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Create all PCollections from the same pipeline object; reuse one pipeline variable across cells.","Move every transform onto a single pipeline: pc2 = pipeline | 'Step2' >> ... instead of pipeline2 | ....","If you accidentally re-created the pipeline, re-run the earlier cells so all PCollections bind to the newest pipeline object.","Call ib.collect once per pipeline, or merge pipelines if they are logically one job.","Ensure you didn't pass a PCollection from a pipeline built with a different runner."],"exampleFix":"// 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)","handlingStrategy":"validation","validationCode":"def same_pipeline(pcolls): return len({p.pipeline for p in pcolls}) == 1; assert same_pipeline(pcolls)","typeGuard":"def belongs_to(pcoll, pipeline): return pcoll.pipeline is pipeline","tryCatchPattern":"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","preventionTips":["Use a single beam.Pipeline() per notebook workflow and reuse it.","Avoid re-running cells that instantiate beam.Pipeline() without re-running dependent cells.","Check set(p.pipeline for p in pcolls) before collecting.","Never mix pcolls from different runners or sessions."],"tags":["python","apache-beam","interactive","value-error","pipelines"],"backgroundTag":"invalid-argument-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}