apache/beam · error · ValueError

Unexpected output element type

Error message

Unexpected output element type %s

What it means

In _write_outputs, outgoing streams are partitioned into Elements.Data and Elements.Timers protobuf sub-messages. A stream object of any other type triggers ValueError 'Unexpected output element type', protecting the data plane from writing malformed output records.

Solutions

1. Verify all writer/transform code emits proper Elements.Data or Elements.Timers messages.
2. Align runner and SDK worker Beam versions to avoid protobuf incompatibilities.
3. Remove or fix any monkey-patching of data plane internals in the worker.
4. Reproduce with a minimal pipeline to identify which transform produces the offending output stream.

Example fix

# before
return SomeCustomWrapper(data)

# after
return beam_fn_api_pb2.Elements.Data(transform_id=transform_id, data=data)

def is_valid_output_stream(s) -> bool:
    import apache_beam.runners.worker.data_plane as dp
    from apache_beam.portability.api import beam_fn_api_pb2
    return isinstance(s, (beam_fn_api_pb2.Elements.Data, beam_fn_api_pb2.Elements.Timers))

When it happens

Trigger: A transform's output writer produces a stream object that is neither beam_fn_api_pb2.Elements.Data nor Elements.Timers when the data plane flushes outputs.

Common situations: Custom sinks/writers or patched Beam internals returning wrong stream types; SDK/runner version mismatch producing incompatible protobuf wrappers; buggy subclassing of data plane output logic.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• Cannot provide because is not a subclass of
• A sink must inherit iobase.Sink, iobase.NativeSink, or be a…
• append() only accepts DeferredDataFrame instances, received
• append() only accepts DeferredSeries instances, received
• bad type: , want

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/worker/data_plane.py:711

               len(streams) <= 100):
          data_or_timer = self._to_send.get_nowait()
          total_size_bytes += data_or_timer.ByteSize()
          streams.append(data_or_timer)
      except queue.Empty:
        pass
      if streams[-1] is self._WRITES_FINISHED:
        stream_done = True
        streams.pop()
      if streams:
        data_stream = []
        timer_stream = []
        for stream in streams:
          if isinstance(stream, beam_fn_api_pb2.Elements.Timers):
            timer_stream.append(stream)
          elif isinstance(stream, beam_fn_api_pb2.Elements.Data):
            data_stream.append(stream)
          else:
            raise ValueError('Unexpected output element type %s' % type(stream))
        yield beam_fn_api_pb2.Elements(data=data_stream, timers=timer_stream)

  def _get_element_size_bytes(self, element):
    # type: (Union[beam_fn_api_pb2.Elements.Data, beam_fn_api_pb2.Elements.Timers]) -> int
    if isinstance(element, beam_fn_api_pb2.Elements.Data):
      return len(element.data)
    elif isinstance(element, beam_fn_api_pb2.Elements.Timers):
      return len(element.timers)
    else:
      return 0

  def _read_inputs(self, elements_iterator):
    # type: (Iterable[beam_fn_api_pb2.Elements]) -> None

    next_discard_log_time = 0  # type: float

    def _put_queue(instruction_id, element):
      # type: (str, Union[beam_fn_api_pb2.Elements.Data, beam_fn_api_pb2.Elements.Timers]) -> None

View on GitHub (pinned to 12126d8942)