apache/beam · error · TypeError

Either {self.class.name}.process_batch() must have a

Error message

Either {self.__class__.__name__}.process_batch() must have a type annotation on its first parameter, or {self.__class__.__name__} must override get_input_batch_type.

What it means

get_input_batch_type (core.py:898) infers a batched DoFn's input element type from the annotation on process_batch's first parameter. If that parameter has no annotation and the DoFn does not override get_input_batch_type, Beam has no way to know the batch input type and raises this TypeError.

Source

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

    input typehint for the first parameter of ``process_batch``. A Batched DoFn
    may override this method if a dynamic approach is required.

    Args:
      input_element_type: The **element type** of the input PCollection this
        DoFn is being applied to.

    Returns:
      ``None`` if this DoFn cannot accept batches, else a Beam typehint or
      a native Python typehint.
    """
    if not self._process_batch_defined:
      return None
    input_type = list(
        inspect.signature(self.process_batch).parameters.values())[0].annotation
    if input_type == inspect.Signature.empty:
      # TODO(https://github.com/apache/beam/issues/21652): Consider supporting
      # an alternative (dynamic?) approach for declaring input type
      raise TypeError(
          f"Either {self.__class__.__name__}.process_batch() must have a type "
          f"annotation on its first parameter, or {self.__class__.__name__} "
          "must override get_input_batch_type.")
    return input_type

  def _get_input_batch_type_normalized(self, input_element_type):
    return typehints.native_type_compatibility.convert_to_beam_type(
        self.get_input_batch_type(input_element_type))

  def _get_output_batch_type_normalized(self, input_element_type):
    return typehints.native_type_compatibility.convert_to_beam_type(
        self.get_output_batch_type(input_element_type))

  @staticmethod
  def _get_element_type_from_return_annotation(method, input_type):
    return_type = inspect.signature(method).return_annotation
    if return_type == inspect.Signature.empty:
      # output type not annotated, try to infer it

View on GitHub (pinned to 12126d8942)

Solutions

Add a type annotation to process_batch's first parameter, e.g. def process_batch(self, batch: List[MyType]).
Override get_input_batch_type in the DoFn class to return the element type explicitly.
Annotate with a batched type like pa.Table or List[T] depending on the batch framework.

Example fix

# before
class MyDoFn(DoFn):
    def process_batch(self, batch):
        ...

# after
class MyDoFn(DoFn):
    def process_batch(self, batch: List[MyType]):
        ...

Defensive patterns

Strategy: validation

Validate before calling

import inspect
def check_process_batch_annotated(cls):
    if hasattr(cls, 'process_batch') and hasattr(type(cls).get_input_batch_type, '__func__'):
        sig = inspect.signature(cls.process_batch)
        first = list(sig.parameters.values())[0]
        if first.annotation is inspect.Signature.empty:
            raise TypeError(f'{cls.__class__.__name__}.process_batch first param needs a type annotation')

Type guard

def batch_input_annotated(cls) -> bool:
    import inspect
    params = list(inspect.signature(cls.process_batch).parameters.values())
    return bool(params) and params[0].annotation is not inspect.Signature.empty

Try / catch

try:
    out = dofn.get_input_batch_type()
except TypeError as e:
    if 'get_input_batch_type' in str(e):
        raise TypeError('Annotate process_batch(self, batch: List[T]) or override get_input_batch_type') from e
    raise

Prevention

Always annotate the first process_batch parameter (e.g. List[T], pa.Table, pd.DataFrame)
Or override get_input_batch_type in the DoFn class
Lint DoFn classes with a custom flake8 plugin requiring return/param annotations on process_batch

When it happens

Trigger: Defining `def process_batch(self, batch):` without any type annotation on `batch` in a DoFn used with batching, without overriding get_input_batch_type.

Common situations: Old-style DoFns written before annotations; batch conversion helpers (e.g. beam.transforms.batch) applied to DoFns whose process_batch lacks hints.

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/f82be864014d5404. Report an issue: GitHub.