What causes "Use p | apache_beam.dataframe.io.%s" in apache/beam?

Use the Beam idiom: p | beam.dataframe.io.read_csv(path) instead of calling the shim function directly.

apache/beam · error · WontImplementError

Use p | apache_beam.dataframe.io.

Error message

Use p | apache_beam.dataframe.io.%s

What it means

The Beam pandas shim module intercepts pandas top-level read_* functions. Instead of executing them eagerly, it raises WontImplementError telling you to use the deferred Beam pipeline form: p | apache_beam.dataframe.io.<name>, i.e. pipe the PCollection/deferred frame through the corresponding io transform, because eager pandas reads cannot run inside a distributed Beam pipeline.

Solutions

Use the Beam idiom: p | beam.dataframe.io.read_csv(path) instead of calling the shim function directly.
Import the real pandas module (import pandas as pd) if you truly want an eager read outside the pipeline.
Use beam.dataframe.convert.to_pcollection/to_dataframe to move data between PCollection and deferred-dataframe worlds.
Check that sys.modules/imports are not shadowing pandas with the Beam shim.

Example fix

// before
df = pd.read_csv('input.csv')  # Beam shim raises
// after
df = p | beam.dataframe.io.read_csv('input.csv')

Defensive patterns

Strategy: fallback

Validate before calling

import pandas as real_pd  # ensure genuine pandas, not the Beam shim
if hasattr(real_pd, '__beam_shim__'):
    raise RuntimeError('pandas is shadowed by beam shim')

Try / catch

try:
    df = pd.read_csv(path)
except frame_base.WontImplementError:
    df = p | beam.dataframe.io.read_csv(path)

Prevention

Use p | beam.dataframe.io.read_* inside pipelines
Keep plain pandas imports separate from Beam deferred usage
Read Beam dataframe docs before porting pandas code

When it happens

Trigger: Calling pd.read_csv/read_json/read_parquet(...) on the apache_beam.dataframe.pandas_top_level_functions module shim (accessed via __getattr__ for names starting with 'read_'), expecting eager pandas semantics.

Common situations: Copy-pasting standard pandas code into a Beam pipeline; IDE autocompletion resolving to the Beam pandas shim instead of real pandas; mixing 'import pandas as pd' with Beam's deferred module in notebook pipelines.

Understand the failure class

Background: "is deprecated and will be removed" — deprecation warnings for old API names, keywords, and options, and how to migrate before the removal release — this error's family across 29 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/dataframe/pandas_top_level_functions.py:176

  pivot = _call_on_first_arg('pivot')
  pivot_table = _call_on_first_arg('pivot_table')
  set_eng_float_format = _defer_to_pandas('set_eng_float_format')
  show_versions = _defer_to_pandas('show_versions')
  test = frame_base.wont_implement_method(
      pd,
      'test',
      explanation="because it is an internal pandas testing utility.")
  timedelta_range = _defer_to_pandas('timedelta_range')
  to_pickle = frame_base.wont_implement_method(
      pd, 'to_pickle', reason='order-sensitive')
  to_datetime = _defer_to_pandas_maybe_elementwise('to_datetime')
  notna = _call_on_first_arg('notna')

  def __getattr__(self, name):
    if name.startswith('read_'):

      def func(*args, **kwargs):
        raise frame_base.WontImplementError(
            'Use p | apache_beam.dataframe.io.%s' % name)

      return func
    res = getattr(pd, name)
    if _is_top_level_function(res):
      return frame_base.not_implemented_method(name, base_type=pd)
    else:
      return res


pd_wrapper = DeferredPandasModule()

View on GitHub (pinned to 12126d8942)