apache/beam · warning

Input to operation must be rebatched from type to . This is…

Error message

Input to operation {consumer} must be rebatched from type {self.producer_batch_converter.batch_type!r} to {consumer_batch_converter.batch_type!r}.
This is very inefficient, consider re-structuring your pipeline or adding a DoFn to directly convert between these types.

What it means

When a producing operation outputs one batch type (e.g. pandas DataFrame or Arrow record batch) and a downstream consumer expects a different batch type, the runner must rebatch every batch, which is expensive. receive_batch emits InefficientExecutionWarning describing the producer/consumer batch types and suggests re-structuring the pipeline.

Solutions

1. Restructure the pipeline so adjacent operations share the same batch type (e.g. use the same batched representation end to end).
2. Insert a DoFn that directly converts between the two batch types to eliminate generic rebatching.
3. Align batch types by switching one side's input/output types (e.g. with MapBlocks or standard element-wise transforms).
4. If the cost is acceptable, filter or ignore InefficientExecutionWarning.

Example fix

# before: producer emits DataFrame batches, consumer wants Arrow
# after: add an explicit converting DoFn
| BeamJarDebug_MapToArrow()  # DoFn converting DataFrame -> arrow.RecordBatch

# Check adjacent transforms share batch types before chaining
# e.g. ensure producer output type == consumer input type (DataFrame vs arrow.RecordBatch)
assert type(producer_output_batch) == type(consumer_input_batch), 'batch types differ; rebatching will occur'

import warnings
from apache_beam.utils.annotations import InefficientExecutionWarning
with warnings.catch_warnings():
    warnings.simplefilter('error', InefficientExecutionWarning)
    run_pipeline()  # fails fast if rebatching happens

When it happens

Trigger: Chaining operations whose batch converters differ — e.g. writing a pandas-backed DoFn/transform output directly into a transform expecting Arrow record batches — so the runner explodes and re-batches elements between them.

Common situations: Mixing standard and batched DoFns over different batch types (DataFrame vs pyarrow), or combining transforms from different subsystems in one optimized pipeline stage.

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

• A cannot be expanded
• Attempted to invoke timer
• Batch size must be a positive integer
• batch type must be pa.Table or pa.Array
• batch type must be pd.Series or pd.DataFrame

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/worker/operations.py:379

    self.update_counters_finish()

  def receive_batch(self, windowed_batch):
    if self.element_consumers:
      for wv in windowed_batch.as_windowed_values(
          self.producer_batch_converter.explode_batch):
        for consumer in self.element_consumers:
          _cast_to_operation(consumer).process(wv)

    for consumer in self.passthrough_batch_consumers:
      _cast_to_operation(consumer).process_batch(windowed_batch)

    for (consumer_batch_converter,
         consumers) in self.other_batch_consumers.items():
      # Explode and rebatch into the new batch type (ouch!)
      # TODO: Register direct conversions for equivalent batch types

      for consumer in consumers:
        warnings.warn(
            f"Input to operation {consumer} must be rebatched from type "
            f"{self.producer_batch_converter.batch_type!r} to "
            f"{consumer_batch_converter.batch_type!r}.\n"
            "This is very inefficient, consider re-structuring your pipeline "
            "or adding a DoFn to directly convert between these types.",
            InefficientExecutionWarning)
        _cast_to_operation(consumer).process_batch(
            windowed_batch.with_values(
                consumer_batch_converter.produce_batch(
                    self.producer_batch_converter.explode_batch(
                        windowed_batch.values))))

    self.update_counters_batch(windowed_batch)

  def flush(self):
    if not self.has_batch_consumers or not self._batched_elements:
      return

View on GitHub (pinned to 12126d8942)