apache/beam · error · NotImplementedError

corrwith( )

Error message

corrwith(%s)

What it means

Apache Beam's DataFrame API (pandas-on-Beam) raised NotImplementedError after attempting to proxy a corrwith() call to pandas. The self._expr.proxy() call is expected to throw since corrwith with a deferred frame argument is not implemented; this raise is a safety net 'in case something else becomes valid'.

Solutions

1. Materialize one or both frames to pandas (e.g. to_pandas()) and use pandas corrwith directly
2. Compute correlations column-wise using self[col].corr(other_col) inside a Beam transform
3. Track/subscribe to the upstream Beam issue for corrwith support and use a non-deferred pandas object in the meantime

Example fix

// before
corr = df.beam.corrwith(other_df.beam)
// after
import pandas as pd
corr = df.beam.to_pandas().corrwith(other_df.beam.to_pandas())

if isinstance(other, frame_base.DeferredFrame):
    raise TypeError('corrwith with a deferred argument is unsupported; materialize with to_pandas()')

def is_concrete_pandas(obj):
    return isinstance(obj, pd.Series) or isinstance(obj, pd.DataFrame)

try:
    corr = dframe.corrwith(other)
except NotImplementedError:
    corr = dframe.to_pandas().corrwith(other.to_pandas())

When it happens

Trigger: Calling DeferredFrame.corrwith(other) where other is a deferred Beam series/frame, with axis/drop/method arguments, in apache_beam.dataframe.frames

Common situations: Porting pandas code that computes pairwise column correlations between two distributed dataframes to Beam pipelines

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• Assigning an index is not yet supported. Consider using…
• by
• concat(ignore_index)
• concat(levels)
• cross join is not yet implemented…

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

Appendix: source

Thrown at sdks/python/apache_beam/dataframe/frames.py:3133

      self, other = self.align(other, axis=0, join='inner')
      col_names = proxy.index
      other_cols = [other] * len(col_names)
    elif isinstance(other, DeferredDataFrame):
      proxy = self._expr.proxy().corrwith(
          other._expr.proxy(), axis=axis, method=method, drop=drop)
      self, other = self.align(other, axis=0, join='inner')
      col_names = list(
          set(self.columns)
          .intersection(other.columns)
          .intersection(proxy.index))
      other_cols = [other[col_name] for col_name in col_names]
    else:
      # Raise the right error.
      self._expr.proxy().corrwith(other._expr.proxy(), axis=axis, drop=drop,
                                  method=method)

      # Just in case something else becomes valid.
      raise NotImplementedError('corrwith(%s)' % type(other._expr.proxy))

    # Generate expressions to compute the actual correlations.
    corrs = [
        self[col_name].corr(other_col, method)
        for col_name, other_col in zip(col_names, other_cols)]

    # Combine the results
    def fill_dataframe(*args):
      result = proxy.copy(deep=True)
      for col, value in zip(proxy.index, args):
        result[col] = value
      return result
    with expressions.allow_non_parallel_operations(True):
      return frame_base.DeferredFrame.wrap(
        expressions.ComputedExpression(
          'fill_dataframe',
          fill_dataframe,
          [corr._expr for corr in corrs],

View on GitHub (pinned to 12126d8942)