apache/beam · error · TypeError

Received from DoFn that was expected to produce a batch.

Error message

Received {type(result).__name__} from DoFn that was expected to produce a batch.

What it means

Beam's batched-DoFn path (_verify_batch_output) expects the process_batch method to return a plain batch (e.g. a list/ndarray of elements), not a WindowedValue or TimestampedValue. Receiving one means the DoFn mixed the per-element output protocol with the batch output protocol, so Beam raises TypeError.

Solutions

Return the raw batch (list/array of values) from process_batch without wrapping in WindowedValue or TimestampedValue.
Move windowing/timestamp handling out of the batched DoFn; the runner applies windowing at the receiver level.
If per-element timestamps are needed, use the standard (non-batched) process() path instead.

Example fix

// before
def process_batch(self, batch):
    return [WindowedValue(x, timestamp, window) for x in batch]
// after
def process_batch(self, batch):
    return [x * 2 for x in batch]  # plain batch, no WindowedValue wrapper

Defensive patterns

Strategy: type-guard

Validate before calling

def _assert_raw_batch(result):
    assert not isinstance(result, (WindowedValue, TimestampedValue)), 'process_batch must return a raw batch'

Type guard

def is_raw_batch(r):
    return not isinstance(r, (WindowedValue, TimestampedValue))

Try / catch

try:
    batch_out = fn.process_batch(batch)
except TypeError as e:
    if 'expected to produce a batch' in str(e):
        raise ValueError('Unwrap WindowedValue/TimestampedValue in process_batch return') from e
    raise

Prevention

Keep batched DoFns free of WindowedValue/TimestampedValue wrappers; let the runner handle windowing.
Write a unit test asserting process_batch returns a plain container type.
Do not copy output code from element-wise DoFns into batched ones.

When it happens

Trigger: A DoFn implementing RunInBatchDoFn/process_batch returns WindowedValue(...) or TimestampedValue(...) instead of a raw batch container; or a batched DoFn reuses code from a standard element-wise DoFn that wraps outputs.

Common situations: Migrating an existing per-element DoFn to batched processing while keeping windowing wrappers around results; copy-pasting output construction from the standard DoFn path.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/common.py:1848

    if tag is None:
      self.main_receivers.receive(windowed_value)
    else:
      self.tagged_receivers[tag].receive(windowed_value)

  def _write_batch_to_tag(self, tag, windowed_batch, watermark_estimator):
    if watermark_estimator is not None:
      for timestamp in windowed_batch.timestamps:
        watermark_estimator.observe_timestamp(timestamp)

    if tag is None:
      self.main_receivers.receive_batch(windowed_batch)
    else:
      self.tagged_receivers[tag].receive_batch(windowed_batch)

  def _verify_batch_output(self, result):
    if isinstance(result, (WindowedValue, TimestampedValue)):
      raise TypeError(
          f"Received {type(result).__name__} from DoFn that was "
          "expected to produce a batch.")

  def start_bundle_outputs(self, results):
    """Validate that start_bundle does not output any elements"""
    if results is None:
      return
    raise RuntimeError(
        'Start Bundle should not output any elements but got %s' % results)

  def finish_bundle_outputs(self, results):
    """Dispatch the result of finish_bundle to the appropriate receivers.

    A value wrapped in a TaggedOutput object will be unwrapped and
    then dispatched to the appropriate indexed output.
    """
    if results is None:
      return

View on GitHub (pinned to 12126d8942)