使用Apache Beam WriteToPubSub遇ProtoType无DESCRIPTOR属性错误

使用Apache Beam的WriteToPubSub转换向Google Cloud Pub/Sub发送数据时，出现以下错误：

AttributeError: 'ProtoType' object has no attribute 'DESCRIPTOR'

该错误关联Protocol Buffers，但难以定位根本原因，简化后的代码如下：

def convert_to_pubsub_message(element):
 if element is None:
    return None
 pubsub_data = {"table": element["table"], "payload": element["values"]}
 pubsub_message = {
    "data": json.dumps(pubsub_data).encode("utf-8"),
    "attributes": {}
 }
 print("Pub/Sub Message:", pubsub_message)
 return pubsub_message

with beam.Pipeline(options=options) as p:
# Read the JSON records from GCS or any source
  records = p | "Read JSON records" >> beam.io.ReadFromText(gcs_json_file)
  # Extract payload and table name from each record
  extracted_data = (
    records
    | "Extract Payload and Table" >> beam.ParDo(ExtractPayloadAndTable())
  )
  # Send extracted payloads to Pub/Sub
  message = (
    extracted_data
    | "Convert to Pub/Sub Message" >> beam.Map(convert_to_pubsub_message)
    | "Send Payload to Pub/Sub" >> WriteToPubSub(topic=pubsub_topic)
  )

• 校验protobuf版本兼容性
  Apache Beam对protobuf版本有严格依赖，不同Beam版本对应特定的protobuf版本区间（例如Beam 2.40+适配protobuf 3.20.x系列），版本过高或过低都会引发属性缺失类错误。先卸载现有protobuf，重新安装匹配版本：

  pip uninstall protobuf -y
  pip install protobuf==3.20.3
  

• 调整消息构造方式，避免字典转proto冲突
  当前代码手动构造包含data和attributes的字典，但WriteToPubSub在部分运行模式（如Dataflow）下，默认期望接收Pub/Sub原生PubsubMessage proto对象而非字典。可以直接返回序列化后的字节数据，让WriteToPubSub自动封装消息结构：
  修改convert_to_pubsub_message函数：

  def convert_to_pubsub_message(element):
      if element is None:
          return None
      pubsub_data = {"table": element["table"], "payload": element["values"]}
      # 直接返回序列化字节，无需手动构造消息字典
      return json.dumps(pubsub_data).encode("utf-8")
  

• 排查Beam SDK及依赖冲突
  确保使用Apache Beam SDK的官方稳定版本，避免预发布版本。同时通过pip freeze查看所有依赖包版本，排查是否有第三方库强制修改了protobuf版本，导致和Beam依赖冲突。

• 统一运行环境的protobuf配置
  若在Dataflow上运行，需确保作业使用的镜像包含正确版本的protobuf：使用自定义镜像时明确指定protobuf版本；使用默认镜像时，选择和本地开发环境版本匹配的Dataflow运行时版本。

内容的提问来源于stack exchange，提问作者Deepak Singh Rajput