apache/beam · error · NotImplementedError

DoFn uses unsupported DoFn param ElementParam.

Error message

DoFn {self.do_fn!r} uses unsupported DoFn param ElementParam.

What it means

process_batch (the DoFn batch API) does not support the per-element DoFn.ElementParam because batch methods receive a whole element sequence; the typehint assumption of the first parameter would break. _validate_process_batch raises NotImplementedError when ElementParam is declared.

Solutions

Remove the ElementParam; the batch method already receives the elements collection itself
Keep the DoFn as a regular process() if per-element access semantics are required
Access element content from the batch parameter (e.g. a list of elements) instead

Example fix

// before
def process_batch(self, els, el=DoFn.ElementParam):
// after
def process_batch(self, els):
  for el in els: ...

Defensive patterns

Strategy: fallback

Validate before calling

import inspect
from apache_beam.transforms.core import DoFn
bad = [d for d in inspect.signature(MyDoFn.process_batch).values if d.default is DoFn.ElementParam]
assert not bad

Prevention

Remember batch methods receive the element collection, not a single element
Avoid mixing per-element params with process_batch

When it happens

Trigger: A DoFn defining process_batch with a parameter defaulted to core.DoFn.ElementParam, validated during DoFnSignature/DoFnInvoker setup via _validate.

Common situations: Migrating a per-element process() to process_batch() without dropping ElementParam; writing a batch DoFn by analogy with per-key params.

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

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

Appendix: source

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

    # type: () -> None

    """Validate that none of the DoFnParameters are repeated in the function
    """
    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

View on GitHub (pinned to 12126d8942)