apache/beam · error · TypeError

{pcoll} is not an apache_beam.pvalue.PCollection.

Error message

{pcoll} is not an apache_beam.pvalue.PCollection.

What it means

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.

Source

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

    # Run the pipeline and bring the PCollection into memory as a Dataframe.
    in_memory_square = head(square, n=5)
    
    # Run the pipeline and get the raw list of elements.
    raw_squares = collect(square, n=5, raw_records=True)
  """
  if len(pcolls) == 0:
    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, '

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass a beam.pvalue.PCollection (the object returned by applying a PTransform to a pipeline).
2. If you have a pandas DataFrame, convert it with apache_beam.dataframes.convert.to_pcollection first, or use ib.transform to keep it deferred.
3. If you have python values, create a PCollection: beam.Create(values) on your interactive pipeline, then collect it.
4. If you passed a list, ensure every element is itself a PCollection/deferred DataFrame; flatten literals into a Create source.
5. 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.

Example fix

// 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)

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))

def is_pcollection(x): import apache_beam as beam; return isinstance(x, beam.pvalue.PCollection)

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

When it happens

Trigger: 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).

Common situations: 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.

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

• pipeline should either be a %s or %s, while %s is given
• A cluster_identifier should be Optional[Union[str, beam.Pipe
• The given pcoll %s is not a dict, an iterable or a PCollecti
• The given pcoll {pcoll_container} is not a dict, an iterable
• Unable to convert objects of type %s to a PCollection

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