如何在Python中处理GCS桶中多JSON对象文件并发布到Pub/Sub

问题原因 #

你的代码运行失败是因为fruit.json采用JSON Lines格式（每行一个独立JSON对象），而非标准的JSON数组。直接用json.loads()解析整个文件内容会触发语法错误——整个文本没有数组包裹、对象间也无逗号分隔，不符合JSON语法规范。

修改后的完整代码 #

import json
from google.cloud import storage
from google.cloud import pubsub_v1

# 初始化GCS客户端
storage_client = storage.Client()
bucket = storage_client.get_bucket('my-buckket_main1')
blob = bucket.blob('fruit.json')

# 初始化Pub/Sub发布客户端（替换为你的项目ID和主题名）
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path('你的GCP项目ID', '你的Pub/Sub主题名称')

# 读取文件内容并逐行处理
content = blob.download_as_text()
for line in content.splitlines():
    if line.strip():  # 跳过空行
        try:
            fruit_obj = json.loads(line)
            # 打印验证解析结果
            print(fruit_obj["fruit"], fruit_obj["size"], fruit_obj["color"])
            
            # 将JSON对象转为字符串，编码为字节后发布到Pub/Sub
            message_str = json.dumps(fruit_obj)
            data = message_str.encode("utf-8")
            future = publisher.publish(topic_path, data=data)
            print(f"已发布消息ID: {future.result()}")
        except json.JSONDecodeError as e:
            print(f"解析JSON行失败: {e}")

关键修改说明 #

• 改用download_as_text()获取文件内容，比字节流更易处理文本行
• 拆分内容为单行，逐行独立解析JSON对象，适配JSON Lines格式
• 补充Pub/Sub发布逻辑：将JSON对象转为字符串后编码为字节（Pub/Sub要求的消息格式）
• 增加空行过滤和异常捕获，避免因空行或格式错误导致程序中断

内容的提问来源于stack exchange，提问作者user3839347