apache/beam · error · TypeError

Failed to find a BatchConverter for the input types of DoFn

Error message

Failed to find a BatchConverter for the input types of DoFn {self.fn!r} (element_type={input_element_type!r}, batch_type={input_batch_type!r}).

What it means

After resolving the DoFn's input batch type, Beam asks BatchConverter.from_typehints to build a converter between element_type and batch_type. If no registered converter supports that (element_type, batch_type) pair, the underlying TypeError is re-raised with this message.

Solutions

Align element_type and batch_type with a registered converter pair (e.g. element_type=np.int64 with batch_type=np.ndarray, or element type backed by pandas DataFrame).
Register a custom BatchConverter via BatchConverter registry (register_batch_converter) for your types.
Install missing optional dependencies (pandas, pyarrow) that provide built-in converters.
Override DoFn.infer_input_type / adjust annotations so the pair is supported.
Example fix: annotate batch as numpy.ndarray and element as np.int64 instead of custom class.

Example fix

// before
class MyDoFn(beam.DoFn):
    def process_batch(self, batch: MyCustomBatchType): ...
// after
class MyDoFn(beam.DoFn):
    def process_batch(self, batch: numpy.ndarray): ...
    def process(self, element: numpy.int64): ...

Defensive patterns

Strategy: validation

Validate before calling

try:
    BatchConverter.from_typehints(element_type=el_t, batch_type=batch_t)
except TypeError:
    raise TypeError('no BatchConverter registered for this (element, batch) pair')

Type guard

def has_batch_converter(el_t, batch_t) -> bool:
    try:
        BatchConverter.from_typehints(element_type=el_t, batch_type=batch_t)
        return True
    except TypeError:
        return False

Try / catch

try:
    run_pipeline_with_batch_dofn()
except TypeError as e:
    if 'BatchConverter' in str(e):
        register_custom_converter(); run_pipeline_with_batch_dofn()
    else:
        raise

Prevention

Stick to supported type pairs (numpy/pandas/arrow).
Install pandas and pyarrow when using batch transforms.
Register custom BatchConverters for domain-specific types.

When it happens

Trigger: Using a DoFn with process_batch where the declared element type and batch type pair has no registered BatchConverter (e.g. element_type=SomeCustomClass, batch_type=pandas.DataFrame), via a ParDo with batch support enabled.

Common situations: Using custom or unsupported element types with batch DoFns; mismatched pandas/numpy/arrow types (e.g. element dict vs batch np.ndarray without a converter); missing optional deps like pandas or pyarrow so the registry lacks converters.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

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

Appendix: source

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

  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:
      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

View on GitHub (pinned to 12126d8942)