apache/beam · error · ValueError

Write transform cannot be constructed for the given proto %r

Error message

Write transform cannot be constructed for the given proto %r

What it means

Write.from_runner_api_parameter reconstructs a Pub/Sub Write transform from its runner API proto. It only supports protos whose spec URN is composites.PUBSUB_WRITE; any other transform proto raises ValueError. This is an internal deserialization guard exercised when a pipeline is deserialized (e.g. portable/runner-side expansion), not a typical user-facing construction path.

Source

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

      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)

  @staticmethod
  @ptransform.PTransform.register_urn(
      common_urns.composites.PUBSUB_WRITE.urn,
      beam_runner_api_pb2.PubSubWritePayload)
  def from_runner_api_parameter(
      ptransform: Any,
      payload: beam_runner_api_pb2.PubSubWritePayload,
      unused_context: PipelineContext,
  ) -> "Write":
    if ptransform.spec.urn != common_urns.composites.PUBSUB_WRITE.urn:
      raise ValueError(
          'Write transform cannot be constructed for the given proto %r',
          ptransform)

    if not payload.topic:
      raise NotImplementedError(
          "from_runner_api_parameter does not "
          "handle empty or None topic")

    # Importing locally to prevent circular dependencies.
    from apache_beam.io.gcp.pubsub import _PubSubSink
    sink = _PubSubSink(
        topic=payload.topic,
        id_label=payload.id_attribute or None,
        timestamp_attribute=payload.timestamp_attribute or None)
    return Write(sink)


class WriteImpl(ptransform.PTransform):

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify the pipeline proto was produced by the same Beam SDK version and that the transform URN is composites.PUBSUB_WRITE.
2. Re-generate the pipeline proto from a fresh pipeline construction instead of reusing a cached/saved pipeline definition.
3. Align SDK versions between the pipeline authoring environment and the expansion/deserialization service.
4. If custom, implement/lookup the correct PTransformProvider for your URN.

Example fix

// before: reusing a stale serialized pipeline proto from Beam 2.20 against a 2.50 runner service
// after: rebuild the pipeline with matching versions, or pin both sides to the same apache-beam version

if transform.spec.urn != 'beam:composites:pubsub_write:v1':
    raise ValueError('unsupported URN: %s' % transform.spec.urn)

def is_pubsub_write_proto(ptransform) -> bool:
    return getattr(getattr(ptransform, 'spec', None), 'urn', None) == 'beam:composites:pubsub_write:v1'

try:
    write = Write.from_runner_api_parameter(ptransform, payload, ctx)
except ValueError as e:
    log.error('Cannot deserialize transform %s: %s', ptransform.spec.urn, e)

When it happens

Trigger: Deserializing a pipeline component whose PTransform proto carries a payload for this writer but has a spec URN other than beams:composites:pubsub_write:v1 - typically a malformed or hand-crafted runner API pipeline proto, or a version/URN mismatch.

Common situations: Portable pipeline deserialization with mismatched Beam SDK versions; corrupted or manually edited pipeline JSON/proto; custom runners replaying protos they altered.

Related errors

• from_runner_api_parameter does not handle empty or None topi
• Unknown type tag %x
• Expected 0 or 1, got %s
• Could not find enum descriptor: {full_name}
• Faled loading session: expected dict, got {}

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