apache/beam · error · ValueError

A sink must inherit iobase.Sink, iobase.NativeSink, or be a

Error message

A sink must inherit iobase.Sink, iobase.NativeSink, or be a PTransform. Received : %r

What it means

apache_beam.io.iobase.Write.expand validates that the sink passed to WriteToTransform is either an iobase.Sink, a NativeSink, or a PTransform. Any other object (a class instead of instance, a string, a connector config object) cannot be written to and raises ValueError. This guards the two supported write paths: custom Sink (wrapped in WriteImpl) and composite PTransform sinks.

Source

Thrown at sdks/python/apache_beam/io/iobase.py:1127

    self.sink = sink

  def display_data(self):
    return {'sink': self.sink.__class__, 'sink_dd': self.sink}

  def expand(self, pcoll):
    # Importing locally to prevent circular dependencies.
    from apache_beam.io.gcp.pubsub import _PubSubSink
    if isinstance(self.sink, _PubSubSink):
      # TODO(BEAM-27443): Remove the need for special casing here.
      return pvalue.PDone(pcoll.pipeline)
    elif isinstance(self.sink, Sink):
      # A custom sink
      return pcoll | WriteImpl(self.sink)
    elif isinstance(self.sink, ptransform.PTransform):
      # This allows "composite" sinks to be used like non-composite ones.
      return pcoll | self.sink
    else:
      raise ValueError(
          'A sink must inherit iobase.Sink, iobase.NativeSink, '
          'or be a PTransform. Received : %r' % self.sink)

  def to_runner_api_parameter(
      self,
      context: PipelineContext,
  ) -> tuple[str, Any]:
    # TODO(BEAM-27443): Remove the need for special casing here.
    # Importing locally to prevent circular dependencies.
    from apache_beam.io.gcp.pubsub import _PubSubSink
    if isinstance(self.sink, _PubSubSink):
      payload = beam_runner_api_pb2.PubSubWritePayload(
          topic=self.sink.full_topic,
          id_attribute=self.sink.id_label,
          timestamp_attribute=self.sink.timestamp_attribute)
      return (common_urns.composites.PUBSUB_WRITE.urn, payload)
    else:
      return super().to_runner_api_parameter(context)

View on GitHub (pinned to 12126d8942)

Solutions

  1. Instantiate the sink: pass MySink(...) (an instance) rather than MySink.
  2. Ensure the sink class inherits from apache_beam.io.iobase.Sink (or NativeSink) and implements write/build_writer.
  3. If the sink is a composite write, make it a PTransform subclass so it is accepted directly.
  4. Prefer the connector's public write transform (e.g. WriteToText, WriteToBigQuery) instead of building WriteToTransform manually.

Example fix

# before
result | WriteToTransform(WriteToText)  # class passed, not instance
# after
result | WriteToTransform(WriteToText('/tmp/out'))  # sink instance
Defensive patterns

Strategy: type-guard

Validate before calling

from apache_beam.io import iobase
assert isinstance(sink, (iobase.Sink, iobase.NativeSink)) or isinstance(sink, beam.transforms.ptransform.PTransform)

Type guard

def is_valid_sink(sink) -> bool:
    from apache_beam.io import iobase
    from apache_beam.transforms import ptransform
    return isinstance(sink, (iobase.Sink, iobase.NativeSink, ptransform.PTransform))

Try / catch

try:
    expanded = WriteToTransform(sink).expand(pcoll)
except ValueError as e:
    if 'A sink must inherit' in str(e): log.error('Pass a Sink/PTransform instance, got %r', sink)

Prevention

When it happens

Trigger: Calling WriteToTransform(sink).expand(pcoll) where sink is not an instance of iobase.Sink, iobase.NativeSink, or ptransform.PTransform - e.g. passing the sink class instead of an instance, or passing a connector-specific config/options object instead of the sink itself.

Common situations: Writing ReadAllResults with a hand-built WriteToTransform; copying code where the sink class name was passed rather than an instance; mixing up a custom IO's config object with its Sink implementation; typos in constructor calls.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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