apache/beam · error · NotImplementedError

DoFn has unsupported per-key DoFn param . Per-key DoFn…

Error message

DoFn {self.do_fn!r} has unsupported per-key DoFn param {d}. Per-key DoFn params are not yet supported for process_batch (https://github.com/apache/beam/issues/21653).

What it means

process_batch does not yet support per-key DoFn parameters (KeyParam, StateParam, TimerParam); these require the per-element/per-key invocation model. _validate_process_batch raises NotImplementedError with a link to Beam issue 21653 tracking the feature.

Solutions

1. Keep process() for stateful/per-key DoFns; batching and state are currently incompatible
2. Move state/timer access into a separate process() DoFn in the pipeline stage
3. Track/upgrade against Beam issue 21653 for when support lands

Example fix

// before
def process_batch(self, els, state=DoFn.StateParam(Spec)):
// after
def process(self, el, state=DoFn.StateParam(Spec)):

import inspect
from apache_beam.transforms.core import DoFn
PER_KEY = (DoFn.KeyParam, DoFn.StateParam, DoFn.TimerParam)
bad = [p for p in inspect.signature(MyDoFn.process_batch).values if p.default in PER_KEY]
assert not bad, 'process_batch cannot use per-key params'

When it happens

Trigger: Declaring process_batch on a stateful DoFn or with core.DoFn.KeyParam/StateParam/TimerParam defaults; validation runs at DoFnSignature creation (_validate).

Common situations: Combining batch processing with stateful/timer logic; converting an existing stateful process() to process_batch hoping for batching performance.

Related errors

• DoFn %r has multiple StateSpecs with the same name
• Input elements to the transform
• Input elements to the transform
• Input value to a stateful DoFn or KeyParam must be a KV…
• A BigQuery table or a query must be specified

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/common.py:365

    self._check_duplicate_dofn_params(self.process_method)

  def _validate_process_batch(self):
    # type: () -> None
    self._check_duplicate_dofn_params(self.process_batch_method)

    for d in self.process_batch_method.defaults:
      if not isinstance(d, core._DoFnParam):
        continue

      # Helpful errors for params which will be supported in the future
      if d == (core.DoFn.ElementParam):
        # We currently assume we can just get the typehint from the first
        # parameter. ElementParam breaks this assumption
        raise NotImplementedError(
            f"DoFn {self.do_fn!r} uses unsupported DoFn param ElementParam.")

      if d in (core.DoFn.KeyParam, core.DoFn.StateParam, core.DoFn.TimerParam):
        raise NotImplementedError(
            f"DoFn {self.do_fn!r} has unsupported per-key DoFn param {d}. "
            "Per-key DoFn params are not yet supported for process_batch "
            "(https://github.com/apache/beam/issues/21653).")

      # Fallback to catch anything not explicitly supported
      if not d in (core.DoFn.WindowParam,
                   core.DoFn.TimestampParam,
                   core.DoFn.PaneInfoParam):
        raise ValueError(
            f"DoFn {self.do_fn!r} has unsupported process_batch "
            f"method parameter {d}")

  def _validate_bundle_method(self, method_wrapper):
    """Validate that none of the DoFnParameters are used in the function
    """
    for param in core.DoFn.DoFnProcessParams:
      if param in method_wrapper.defaults:
        raise ValueError(

View on GitHub (pinned to 12126d8942)