如何合并不同PubSub对象并扁平化写入BigQuery为单行JSON

核心思路 #

把Pub/Sub消息的data、attributes、publishTime三个核心部分拆解并扁平化为顶层字段，生成符合BigQuery Schema的扁平JSON对象，确保所有元素对应单行存储。

完整代码实现 #

import base64
import json
from google.cloud import pubsub_v1

def flatten_pubsub_message(message: pubsub_v1.types.PubsubMessage) -> dict:
    # 1. 解码并解析data字段
    decoded_data = base64.b64decode(message.data).decode("utf-8")
    data_json = json.loads(decoded_data)
    
    # 2. 扁平化attributes字典（直接转为顶层字段）
    attributes_flat = message.attributes
    
    # 3. 格式化publishTime为BigQuery兼容格式
    publish_time = message.publish_time.isoformat()
    
    # 4. 合并所有字段为扁平字典
    flattened_message = {**data_json, **attributes_flat, "publishTime": publish_time}
    
    return flattened_message

代码说明 #

• data字段处理：Pub/Sub消息的data是base64编码字节流，先解码为UTF-8字符串，再解析为JSON对象，直接展开为顶层字段。
• attributes扁平化：attributes本身是键值对字典，直接将所有键值对合并到顶层，避免嵌套结构。
• publishTime适配：将消息自带的publish_time（datetime类型）转为ISO 8601格式字符串，匹配BigQuery的TIMESTAMP类型要求。
• 字段合并：通过字典解包语法将data、attributes、publishTime的所有字段整合为一个无嵌套的扁平字典。

BigQuery Schema适配 #

结合你提供的示例字段，对应的BigQuery Schema可定义为：

[
    {"name": "data[0]", "type": "STRING", "mode": "NULLABLE"},
    {"name": "data[1]", "type": "STRING", "mode": "NULLABLE"},
    {"name": "attr[0]", "type": "STRING", "mode": "NULLABLE"},
    {"name": "key", "type": "STRING", "mode": "NULLABLE"},
    {"name": "publishTime", "type": "TIMESTAMP", "mode": "NULLABLE"}
]

写入BigQuery示例 #

使用BigQuery客户端库完成数据写入：

from google.cloud import bigquery

def write_to_bigquery(flattened_msg: dict, table_id: str):
    client = bigquery.Client()
    errors = client.insert_rows_json(table_id, [flattened_msg])
    if not errors:
        print("消息已成功写入BigQuery")
    else:
        print(f"写入失败: {errors}")

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