apache/beam · error · ValueError

DoFn %r has duplicate

Error message

DoFn %r has duplicate %s method parameters: %s.

What it means

get_dofn_specs inspects a stateful DoFn's process/start_bundle/finish_bundle (and timer) signatures for _DoFnParam defaults (e.g. DoFn.StateParam, DoFn.TimerParam, DoFn.TimestampParam). If the same parameter (same param_id) appears more than once in one method's defaults, the spec set would be inconsistent, so a ValueError naming the DoFn, method, and duplicate param ids is raised during pipeline validation.

Solutions

1. Reference each state/timer spec at most once per method signature and reuse the value inside the method body
2. If you need the same state in multiple places, bind the parameter once and pass the handle around
3. Fix duplicated parameter defaults so each _DoFnParam in the signature has a distinct spec

Example fix

// before
def process(self, item, a=DoFn.StateParam(BAG_SPEC), b=DoFn.StateParam(BAG_SPEC)): ...
// after
def process(self, item, bag=DoFn.StateParam(BAG_SPEC)):
    ...use bag...

from apache_beam.transforms.userstate import get_dofn_specs
# fail fast in tests before submitting the pipeline
get_dofn_specs(MyDoFn)

try:
    validate_stateful_dofn(MyDoFn)
except ValueError as e:
    raise PipelineDefinitionError(f'fix DoFn signature: {e}') from e

When it happens

Trigger: Writing def process(self, item, state=DoFn.StateParam(SPEC), other=DoFn.StateParam(SPEC)) — the same state spec used twice in one method; two params sharing the same underlying spec/param id; duplicate TimerParam/BagStateParam defaults.

Common situations: Copy-pasting a parameter line and forgetting to change the spec; deliberately reading the same state twice in a signature instead of inside the method body; auto-generated DoFn signatures emitting repeated params.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• An unsupported sink was specified
• At least one of --render_port or --render_output must be…
• buffer_sec must be >= 0, got
• Cannot skip negative number of header lines
• change_function must be 'CHANGES' or 'APPENDS', got

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/userstate.py:274

  from apache_beam.runners.common import MethodWrapper
  from apache_beam.transforms.core import _DoFnParam
  from apache_beam.transforms.core import _StateDoFnParam
  from apache_beam.transforms.core import _TimerDoFnParam

  all_state_specs = set()
  all_timer_specs = set()

  # Validate params to process(), start_bundle(), finish_bundle() and to
  # any on_timer callbacks.
  for method_name in dir(dofn):
    if not isinstance(getattr(dofn, method_name, None), types.MethodType):
      continue
    method = MethodWrapper(dofn, method_name)
    param_ids = [
        d.param_id for d in method.defaults if isinstance(d, _DoFnParam)
    ]
    if len(param_ids) != len(set(param_ids)):
      raise ValueError(
          'DoFn %r has duplicate %s method parameters: %s.' %
          (dofn, method_name, param_ids))
    for d in method.defaults:
      if isinstance(d, _StateDoFnParam):
        all_state_specs.add(d.state_spec)
      elif isinstance(d, _TimerDoFnParam):
        all_timer_specs.add(d.timer_spec)

  return all_state_specs, all_timer_specs


def is_stateful_dofn(dofn: 'DoFn') -> bool:
  """Determines whether a given DoFn is a stateful DoFn."""

  # A Stateful DoFn is a DoFn that uses user state or timers.
  all_state_specs, all_timer_specs = get_dofn_specs(dofn)
  return bool(all_state_specs or all_timer_specs)

View on GitHub (pinned to 12126d8942)