{"record":{"id":"14a18a05bf7cb192","repo":"apache/beam","slug":"input-to-operation-consumer-must-be-rebatched-from-type-self","errorCode":null,"errorMessage":"Input to operation {consumer} must be rebatched from type {self.producer_batch_converter.batch_type!r} to {consumer_batch_converter.batch_type!r}.\nThis is very inefficient, consider re-structuring your pipeline or adding a DoFn to directly convert between these types.","messagePattern":"Input to operation (.+?) must be rebatched from type (.+?) to (.+?)\\.\nThis is very inefficient, consider re-structuring your pipeline or adding a DoFn to directly convert between these types\\.","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/python/apache_beam/runners/worker/operations.py","lineNumber":379,"sourceCode":"    self.update_counters_finish()\n\n  def receive_batch(self, windowed_batch):\n    if self.element_consumers:\n      for wv in windowed_batch.as_windowed_values(\n          self.producer_batch_converter.explode_batch):\n        for consumer in self.element_consumers:\n          _cast_to_operation(consumer).process(wv)\n\n    for consumer in self.passthrough_batch_consumers:\n      _cast_to_operation(consumer).process_batch(windowed_batch)\n\n    for (consumer_batch_converter,\n         consumers) in self.other_batch_consumers.items():\n      # Explode and rebatch into the new batch type (ouch!)\n      # TODO: Register direct conversions for equivalent batch types\n\n      for consumer in consumers:\n        warnings.warn(\n            f\"Input to operation {consumer} must be rebatched from type \"\n            f\"{self.producer_batch_converter.batch_type!r} to \"\n            f\"{consumer_batch_converter.batch_type!r}.\\n\"\n            \"This is very inefficient, consider re-structuring your pipeline \"\n            \"or adding a DoFn to directly convert between these types.\",\n            InefficientExecutionWarning)\n        _cast_to_operation(consumer).process_batch(\n            windowed_batch.with_values(\n                consumer_batch_converter.produce_batch(\n                    self.producer_batch_converter.explode_batch(\n                        windowed_batch.values))))\n\n    self.update_counters_batch(windowed_batch)\n\n  def flush(self):\n    if not self.has_batch_consumers or not self._batched_elements:\n      return\n","sourceCodeStart":361,"sourceCodeEnd":397,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/operations.py#L361-L397","documentation":"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.","triggerScenarios":"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.","commonSituations":"Mixing standard and batched DoFns over different batch types (DataFrame vs pyarrow), or combining transforms from different subsystems in one optimized pipeline stage.","solutions":["Restructure the pipeline so adjacent operations share the same batch type (e.g. use the same batched representation end to end).","Insert a DoFn that directly converts between the two batch types to eliminate generic rebatching.","Align batch types by switching one side's input/output types (e.g. with MapBlocks or standard element-wise transforms).","If the cost is acceptable, filter or ignore InefficientExecutionWarning."],"exampleFix":"# before: producer emits DataFrame batches, consumer wants Arrow\n# after: add an explicit converting DoFn\n| BeamJarDebug_MapToArrow()  # DoFn converting DataFrame -> arrow.RecordBatch","handlingStrategy":"validation","validationCode":"# Check adjacent transforms share batch types before chaining\n# e.g. ensure producer output type == consumer input type (DataFrame vs arrow.RecordBatch)\nassert type(producer_output_batch) == type(consumer_input_batch), 'batch types differ; rebatching will occur'","typeGuard":null,"tryCatchPattern":"import warnings\nfrom apache_beam.utils.annotations import InefficientExecutionWarning\nwith warnings.catch_warnings():\n    warnings.simplefilter('error', InefficientExecutionWarning)\n    run_pipeline()  # fails fast if rebatching happens","preventionTips":["Keep one batch representation (e.g. Arrow record batches) across batched DoFns","Write explicit converting DoFns between pandas and Arrow sections","Treat InefficientExecutionWarning as an error in tests to catch pipeline hot spots"],"tags":["performance","batching","inefficiency","runner"],"backgroundTag":"incompatible-source-type","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}