{"record":{"id":"7e999bb776b32881","repo":"apache/beam","slug":"pipeline-keyword-required-for-non-deferred-dataframe","errorCode":null,"errorMessage":"Pipeline keyword required for non-deferred dataframe conversion.","messagePattern":"Pipeline keyword required for non-deferred dataframe conversion\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/dataframe/convert.py","lineNumber":227,"sourceCode":"  \"\"\"\n  if not yield_elements in (\"pandas\", \"schemas\"):\n    raise ValueError(\n        \"Invalid value for yield_elements argument, '%s'. \"\n        \"Allowed values are 'pandas' and 'schemas'\" % yield_elements)\n  if label is None:\n    # Attempt to come up with a reasonable, stable label by retrieving the name\n    # of these variables in the calling context.\n    label = 'ToPCollection(%s)' % ', '.join(_var_name(e, 3) for e in dataframes)\n\n  # Support for non-deferred dataframes.\n  deferred_dataframes = []\n  for ix, df in enumerate(dataframes):\n    if isinstance(df, frame_base.DeferredBase):\n      # TODO(robertwb): Maybe extract pipeline object?\n      deferred_dataframes.append(df)\n    elif isinstance(df, (pd.Series, pd.DataFrame)):\n      if pipeline is None:\n        raise ValueError(\n            'Pipeline keyword required for non-deferred dataframe conversion.')\n      deferred = pipeline | '%s_Defer%s' % (label, ix) >> beam.Create([df])\n      deferred_dataframes.append(\n          frame_base.DeferredFrame.wrap(\n              expressions.PlaceholderExpression(df.iloc[:0], deferred)))\n    else:\n      raise TypeError(\n          'Unable to convert objects of type %s to a PCollection' % type(df))\n  dataframes = tuple(deferred_dataframes)\n\n  def extract_input(placeholder):\n    if not isinstance(placeholder._reference, pvalue.PCollection):\n      raise TypeError(\n          'Expression roots must have been created with to_dataframe.')\n    return placeholder._reference\n\n  placeholders = frozenset.union(\n      frozenset(), *[df._expr.placeholders() for df in dataframes])","sourceCodeStart":209,"sourceCodeEnd":245,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/dataframe/convert.py#L209-L245","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nturned = convert.to_pcollection(local_df)\n// after\nwith beam.Pipeline() as p:\n    turned = convert.to_pcollection(local_df, pipeline=p)","handlingStrategy":"validation","validationCode":"import pandas as pd\nfrom apache_beam.dataframe import frame_base\nif pipeline is None and any(isinstance(d, (pd.Series, pd.DataFrame))\n                            and not isinstance(d, frame_base.DeferredBase)\n                            for d in dataframes):\n    raise ValueError('pipeline= is required for concrete pandas inputs')","typeGuard":null,"tryCatchPattern":"try:\n    out = convert.to_pcollection(local_df)\nexcept ValueError as e:\n    if 'Pipeline keyword required' in str(e):\n        with beam.Pipeline() as p:\n            out = convert.to_pcollection(local_df, pipeline=p)\n    else:\n        raise","preventionTips":["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."],"tags":["python","dataframe","pipeline","missing-argument"],"backgroundTag":"missing-required-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}