借助BigQuery订阅将GCP Pub/Sub含JSON数组的消息拆分存入BigQuery多行

AI 小北

2026-7-20

拆分Pub/Sub合并JSON数组到BigQuery多行的方法

完全可行，以下是两种常用的实现方案：

方案一：直接通过BigQuery内置函数拆分（无额外服务成本）

当Pub/Sub消息以JSON数组字符串的形式写入BigQuery后，你可以利用JSON_EXTRACT_ARRAY和UNNEST函数将数组拆分为独立行。

操作步骤：

确保Pub/Sub到BigQuery的订阅将消息内容写入到一个包含字符串类型字段（比如命名为data）的表中。
使用以下SQL查询拆分数组并写入目标表：

INSERT INTO `your-project.your-dataset.target-table` (name, gender)
SELECT
  JSON_EXTRACT_SCALAR(item, '$.name') AS name,
  JSON_EXTRACT_SCALAR(item, '$.gender') AS gender
FROM
  `your-project.your-dataset.source-table`,
  UNNEST(JSON_EXTRACT_ARRAY(data)) AS item

若要实现自动实时拆分，可以在创建Pub/Sub订阅时选择「使用查询转换数据」，将上述SQL作为转换逻辑，直接将拆分后的结果写入目标表，无需手动执行查询。

方案二：用Dataflow预处理后写入（适合复杂场景）

如果需要对消息做额外的清洗、校验或格式转换，可以用Dataflow订阅Pub/Sub消息，在数据流中拆分数组后再写入BigQuery。

Python示例代码：

import apache_beam as beam
import json

# 拆分JSON数组为单个元素
def split_array(message):
    try:
        json_data = json.loads(message.data.decode('utf-8'))
        for item in json_data:
            yield item
    except json.JSONDecodeError:
        # 处理格式错误的消息
        pass

with beam.Pipeline() as pipeline:
    (pipeline
     | "读取Pub/Sub消息" >> beam.io.ReadFromPubSub(subscription="projects/your-project/subscriptions/your-sub")
     | "拆分JSON数组" >> beam.FlatMap(split_array)
     | "写入BigQuery" >> beam.io.WriteToBigQuery(
         table="your-project:your-dataset.target-table",
         schema="name:STRING, gender:STRING",
         write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
     ))