{"record":{"id":"9385e04f42d15124","repo":"apache/beam","slug":"pcoll-is-not-an-apache-beam-pvalue-pcollection","errorCode":null,"errorMessage":"{pcoll} is not an apache_beam.pvalue.PCollection.","messagePattern":"(.+?) is not an apache_beam\\.pvalue\\.PCollection\\.","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/interactive_beam.py","lineNumber":931,"sourceCode":"    # Run the pipeline and bring the PCollection into memory as a Dataframe.\n    in_memory_square = head(square, n=5)\n    \n    # Run the pipeline and get the raw list of elements.\n    raw_squares = collect(square, n=5, raw_records=True)\n  \"\"\"\n  if len(pcolls) == 0:\n    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, '","sourceCodeStart":913,"sourceCodeEnd":949,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/interactive_beam.py#L913-L949","documentation":"The internal helper as_pcollection (used by ib.collect) normalizes each input into (PCollection, element_type). Deferred DataFrames are converted; beam.pvalue.PCollection instances pass through. Anything else triggers this TypeError. The message interpolates the loop variable pcoll rather than the parameter pcoll_or_df — a library bug — so the interpolated text may be wrong; either way the cause is the same: you gave ib.collect() something that is neither a PCollection nor a deferred DataFrame.","triggerScenarios":"ib.collect(5); ib.collect(pipeline.run() result); ib.collect([pcoll, 42]) (list containing a non-PCollection); ib.collect(raw_pandas_df) where the object is pandas and not a beam DeferredBase; ib.collect(list_from_previous_collect).","commonSituations":"Notebooks: collecting a pandas DataFrame directly assuming ib.collect converts it (it only handles beam deferred DataFrames); collecting a list of pipeline values instead of PCollections; passing a materialized python list from a previous collect back into collect; a refactor left a stale variable.","solutions":["Pass a beam.pvalue.PCollection (the object returned by applying a PTransform to a pipeline).","If you have a pandas DataFrame, convert it with apache_beam.dataframes.convert.to_pcollection first, or use ib.transform to keep it deferred.","If you have python values, create a PCollection: beam.Create(values) on your interactive pipeline, then collect it.","If you passed a list, ensure every element is itself a PCollection/deferred DataFrame; flatten literals into a Create source.","Ignore the interpolated text (variable bug) and inspect each element's type with type(x) — the code checks isinstance of beam.pvalue.PCollection or DeferredBase."],"exampleFix":"// before: ib.collect(pandas_df)  # plain pandas DF | // after: from apache_beam.dataframes.convert import to_pcollection; pc, _ = to_pcollection(pandas_df, label='df'); ib.collect(pc)","handlingStrategy":"type-guard","validationCode":"import apache_beam as beam; from apache_beam.dataframes import DeferredBase; def is_collectable(x): return isinstance(x, (beam.pvalue.PCollection, DeferredBase)); assert all(map(is_collectable, items))","typeGuard":"def is_pcollection(x): import apache_beam as beam; return isinstance(x, beam.pvalue.PCollection)","tryCatchPattern":"try: df = ib.collect(pc) | except TypeError as e: print('collect() got non-PCollection input:', e)  # message may show a wrong variable name (library bug); verify types yourself","preventionTips":["Ensure every argument is a PCollection or beam deferred DataFrame/Series.","Convert pandas DataFrames via apache_beam.dataframes.convert.to_pcollection first.","Create literal data with beam.Create on the interactive pipeline.","Verify element types yourself; the error text can be misleading."],"tags":["python","apache-beam","interactive","type-error","argument-validation"],"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"}