apache/beam · error · TypeError

process_batch method on

Error message

process_batch method on {self.fn!r} does not have a return type annoation

What it means

If a DoFn can yield batches (_can_yield_batches) but its process_batch return type annotation cannot be normalized to a batch type (None), Beam raises this TypeError, since it must know the output batch type to build the output converter.

Solutions

Add a return type annotation to process_batch, e.g. def process_batch(self, b: np.ndarray) -> Iterator[np.ndarray].
Ensure the annotation resolves to a concrete batch type (not a bare Iterator without element type).
Avoid annotation-stripping decorators; re-add typing info if wrapped.
If batches are not intended, remove batch-yield logic and use plain process.
Example fix: `def process_batch(self, batch: np.ndarray) -> Iterator[np.ndarray]` instead of no return annotation.

Example fix

// before
class MyDoFn(beam.DoFn):
    def process_batch(self, batch: np.ndarray):
        yield batch * 2
// after
class MyDoFn(beam.DoFn):
    def process_batch(self, batch: np.ndarray) -> Iterator[np.ndarray]:
        yield batch * 2

Defensive patterns

Strategy: validation

Validate before calling

hints = typing.get_type_hints(MyDoFn.process_batch)
if 'return' not in hints:
    raise TypeError('process_batch needs a return type annotation')

Type guard

def process_batch_return_annotated(dofn_cls) -> bool:
    return 'return' in get_type_hints(dofn_cls.process_batch)

Prevention

Always annotate process_batch return types like Iterator[np.ndarray].
Include element types in Iterator/Iterable annotations, not bare forms.
Run a small local pipeline test to surface annotation problems before deployment.

When it happens

Trigger: Defining a batch-yielding DoFn whose process_batch lacks a return type annotation (or whose annotation normalizes to None), then running it through a batch-enabled ParDo which calls _get_output_batch_type_normalized.

Common situations: Forgetting `-> Iterable[...]` / `-> Iterator[np.ndarray]` style return hints on process_batch; decorators stripping annotations; writing process (not batch) docs but yielding batches inadvertently.

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

Appendix: source

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

        # 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:
      output_batch_type = self.fn._get_output_batch_type_normalized(
          input_element_type)
      if output_batch_type is None:
        # TODO: Mention process method in this error
        raise TypeError(
            f"process_batch method on {self.fn!r} does not have "
            "a return type annoation")

      # Generate a batch converter to convert between the output type and the
      # (batch) output type of process_batch
      output_element_type = self.infer_output_type(input_element_type)

      try:
        self.fn.output_batch_converter = BatchConverter.from_typehints(
            element_type=output_element_type, batch_type=output_batch_type)
      except TypeError as e:
        raise TypeError(
            "Failed to find a BatchConverter for the *output* types of DoFn "
            f"{self.fn!r} (element_type={output_element_type!r}, "
            f"batch_type={output_batch_type!r}). Maybe you need to override "
            "DoFn.infer_output_type to set the output element type?") from e
    else:
      self.fn.output_batch_converter = None

View on GitHub (pinned to 12126d8942)