apache/beam · error · NotImplementedError

to_runner_api_parameter not implemented for type

Error message

to_runner_api_parameter not implemented for type

What it means

When Beam serializes a CustomSource transform to the runner API, to_runner_api_parameter dispatches on the wrapped source type: UnboundedSource goes to the base class and PTransform sources delegate to their own method. If the wrapped source is neither, NotImplementedError 'to_runner_api_parameter not implemented for type' is raised, meaning Beam does not know how to translate that source for the pipeline runner.

Source

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

              subscription=self.source.full_subscription,
              timestamp_attribute=self.source.timestamp_attribute,
              with_attributes=self.source.with_attributes,
              id_attribute=self.source.id_label))
    if isinstance(self.source, BoundedSource):
      return (
          common_urns.deprecated_primitives.READ.urn,
          beam_runner_api_pb2.ReadPayload(
              source=self.source.to_runner_api(context),
              is_bounded=beam_runner_api_pb2.IsBounded.BOUNDED
              if self.source.is_bounded() else
              beam_runner_api_pb2.IsBounded.UNBOUNDED))
    # Local import to avoid a circular dependency.
    from apache_beam.io.unbounded_source import UnboundedSource
    if isinstance(self.source, UnboundedSource):
      return super().to_runner_api_parameter(context)
    elif isinstance(self.source, ptransform.PTransform):
      return self.source.to_runner_api_parameter(context)
    raise NotImplementedError(
        "to_runner_api_parameter not "
        "implemented for type")

  @staticmethod
  def from_runner_api_parameter(
      transform: beam_runner_api_pb2.PTransform,
      payload: Union[beam_runner_api_pb2.ReadPayload,
                     beam_runner_api_pb2.PubSubReadPayload],
      context: PipelineContext,
  ) -> "Read":
    if transform.spec.urn == common_urns.composites.PUBSUB_READ.urn:
      assert isinstance(payload, beam_runner_api_pb2.PubSubReadPayload)
      # Importing locally to prevent circular dependencies.
      # TODO(BEAM-27443): Remove the need for this.
      from apache_beam.io.gcp.pubsub import _PubSubSource
      source = _PubSubSource(
          topic=payload.topic or None,
          subscription=payload.subscription or None,

View on GitHub (pinned to 12126d8942)

Solutions

Implement to_runner_api_parameter (and from_runner_api_parameter) on your source class, or make it a proper PTransform so delegation works.
Migrate the custom source to a supported BoundedSource/filebasedsource-based implementation or a built-in Beam connector.
Wrap your reading logic as a PTransform and use that as the source.
Check Beam version compatibility of the third-party io module and upgrade to a release supporting the runner API.
Inspect self.source's actual type/class to confirm why it matches neither branch.

Defensive patterns

Strategy: type-guard

Validate before calling

from apache_beam.io.iobase import CustomSource
from apache_beam.io.unbounded_source import UnboundedSource
import apache_beam as beam
assert isinstance(src, (UnboundedSource, beam.PTransform)), \
    f'unsupported source type for runner API: {type(src)}'

Type guard

def is_runner_api_compatible_source(src):
    from apache_beam.io.unbounded_source import UnboundedSource
    import apache_beam as beam
    return isinstance(src, (UnboundedSource, beam.PTransform))

Try / catch

try:
    pipeline.run()
except NotImplementedError as e:
    if 'to_runner_api_parameter' in str(e):
        raise TypeError(
            'Custom source is not serializable to the runner API; '
            'convert it to a PTransform or supported BoundedSource') from e
    raise

Prevention

Avoid wrapping legacy io.Source subclasses in CustomSource for portable runners.
Prefer built-in Beam connectors or filebasedsource-based implementations.
Register custom to_runner_api_parameter/from_runner_api_parameter for custom transforms.
Pin Beam and connector library versions so source classes inherit the expected base types.

When it happens

Trigger: Using a custom Source object that is neither an UnboundedSource nor a PTransform as the payload of CustomSource (e.g. a legacy io.Source subclass or a hand-rolled source) on a portable/Flink/Dataflow runner; wrapping a plain object by mistake.

Common situations: Migrating old batch Source-based connectors to newer Beam versions with portable pipelines; third-party or in-house sources never updated for the runner API; version mismatch where an expected io class no longer inherits UnboundedSource.

Related errors

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