apache/beam · error · TypeError

process_batch method on

Error message

process_batch method on {self.fn!r} does not have an input type annoation

What it means

During batch DoFn setup, if a DoFn defines a process_batch method but its input batch type annotation cannot be resolved (returns None), Beam raises this TypeError. The annotation is required to build the element↔batch converter.

Solutions

Add a concrete input type annotation to process_batch, e.g. def process_batch(self, batch: np.ndarray).
Ensure annotations are preserved (avoid decorators that strip __annotations__).
Verify with DoFn._get_input_batch_type_normalized that the hint resolves to a batch type.
If batch processing is not intended, remove process_batch and use process instead.
Example fix: `def process_batch(self, batch: pandas.DataFrame)` instead of `def process_batch(self, batch)`.

Example fix

// before
class MyDoFn(beam.DoFn):
    def process_batch(self, batch):
        yield batch.sum()
// after
class MyDoFn(beam.DoFn):
    def process_batch(self, batch: pandas.DataFrame):
        yield batch.sum()

Defensive patterns

Strategy: validation

Validate before calling

hints = typing.get_type_hints(MyDoFn.process_batch)
if 'batch' not in hints or hints['batch'] is None:
    raise TypeError('process_batch needs a concrete input type annotation')

Type guard

def process_batch_annotated(dofn_cls) -> bool:
    return hasattr(dofn_cls, 'process_batch') and bool(get_type_hints(dofn_cls.process_batch).get('batch'))

Prevention

Always annotate process_batch parameters and returns fully.
Avoid decorators that strip __annotations__ from batch methods.
Test batch DoFns locally before running on a runner.

When it happens

Trigger: Defining a DoFn with process_batch but omitting an input type annotation on it (or annotating in a way that normalizes to None), then using it in a ParDo that enables batch processing (e.g. with a runner/transform that calls _get_input_batch_type_normalized).

Common situations: Writing batch-optimized DoFns (e.g. for pandas/arrow) and forgetting type hints on process_batch; hints removed by wrappers/decorators; using older Beam versions where batch support annotations differ.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/core.py:1717

      return self
    else:
      return self.with_exception_handling(
          error_handler=error_handler, **exception_handling_kwargs)

  def default_type_hints(self):
    return self.fn.get_type_hints()

  def infer_output_type(self, input_type):
    return self.fn.infer_output_type(input_type)

  def infer_batch_converters(self, input_element_type):
    # TODO: Test this code (in batch_dofn_test)
    if self.fn._process_batch_defined:
      input_batch_type = self.fn._get_input_batch_type_normalized(
          input_element_type)

      if input_batch_type is None:
        raise TypeError(
            "process_batch method on {self.fn!r} does not have "
            "an input type annoation")

      try:
        # Generate a batch converter to convert between the input type and the
        # (batch) input type of process_batch
        self.fn.input_batch_converter = BatchConverter.from_typehints(
            element_type=input_element_type, batch_type=input_batch_type)
      except TypeError as e:
        raise TypeError(
            "Failed to find a BatchConverter for the input types of DoFn "
            f"{self.fn!r} (element_type={input_element_type!r}, "
            f"batch_type={input_batch_type!r}).") from e

    else:
      self.fn.input_batch_converter = None

    if self.fn._can_yield_batches:

View on GitHub (pinned to 12126d8942)