apache/beam · error · ValueError

Pipeline keyword required for non-deferred dataframe convers

Error message

Pipeline keyword required for non-deferred dataframe conversion.

What it means

convert.to_pcollection can convert non-deferred (concrete pandas) inputs only by inserting them into a pipeline via beam.Create; since a concrete Series/DataFrame carries no pipeline reference, the pipeline keyword is mandatory. Passing only concrete dataframes without pipeline raises this ValueError.

Source

Thrown at sdks/python/apache_beam/dataframe/convert.py:227

  """
  if not yield_elements in ("pandas", "schemas"):
    raise ValueError(
        "Invalid value for yield_elements argument, '%s'. "
        "Allowed values are 'pandas' and 'schemas'" % yield_elements)
  if label is None:
    # Attempt to come up with a reasonable, stable label by retrieving the name
    # of these variables in the calling context.
    label = 'ToPCollection(%s)' % ', '.join(_var_name(e, 3) for e in dataframes)

  # Support for non-deferred dataframes.
  deferred_dataframes = []
  for ix, df in enumerate(dataframes):
    if isinstance(df, frame_base.DeferredBase):
      # TODO(robertwb): Maybe extract pipeline object?
      deferred_dataframes.append(df)
    elif isinstance(df, (pd.Series, pd.DataFrame)):
      if pipeline is None:
        raise ValueError(
            'Pipeline keyword required for non-deferred dataframe conversion.')
      deferred = pipeline | '%s_Defer%s' % (label, ix) >> beam.Create([df])
      deferred_dataframes.append(
          frame_base.DeferredFrame.wrap(
              expressions.PlaceholderExpression(df.iloc[:0], deferred)))
    else:
      raise TypeError(
          'Unable to convert objects of type %s to a PCollection' % type(df))
  dataframes = tuple(deferred_dataframes)

  def extract_input(placeholder):
    if not isinstance(placeholder._reference, pvalue.PCollection):
      raise TypeError(
          'Expression roots must have been created with to_dataframe.')
    return placeholder._reference

  placeholders = frozenset.union(
      frozenset(), *[df._expr.placeholders() for df in dataframes])

View on GitHub (pinned to 12126d8942)

Solutions

Pass pipeline=p (a beam.Pipeline instance) when any input is a concrete pd.DataFrame/pd.Series.
Convert data first to deferred form, or construct the Pipeline before calling to_pcollection.
Reorder code so the Pipeline object exists before the conversion call.
Wrap the call with a check: if pipeline is None and not any deferred inputs, raise a clear local error.

Example fix

// before
turned = convert.to_pcollection(local_df)
// after
with beam.Pipeline() as p:
    turned = convert.to_pcollection(local_df, pipeline=p)

Defensive patterns

Strategy: validation

Validate before calling

import pandas as pd
from apache_beam.dataframe import frame_base
if pipeline is None and any(isinstance(d, (pd.Series, pd.DataFrame))
                            and not isinstance(d, frame_base.DeferredBase)
                            for d in dataframes):
    raise ValueError('pipeline= is required for concrete pandas inputs')

Try / catch

try:
    out = convert.to_pcollection(local_df)
except ValueError as e:
    if 'Pipeline keyword required' in str(e):
        with beam.Pipeline() as p:
            out = convert.to_pcollection(local_df, pipeline=p)
    else:
        raise

Prevention

Create the beam.Pipeline before any dataframe conversion code.
Always pass pipeline= when inputs may include concrete pandas objects.
Distinguish deferred vs concrete inputs in helper APIs.
In notebooks, keep pipeline construction at the top of the workflow.

When it happens

Trigger: Calling convert.to_pcollection(pd.DataFrame(...)) with no pipeline argument; passing a mix where at least one input is concrete pandas while pipeline=None; losing the pipeline variable in refactored code.

Common situations: Notebook conversion of local pandas dataframes into pipelines; tests reusing helpers that previously only got deferred inputs; scripts that build dataframes locally before creating the Pipeline object.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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