如何批量修改GCS中JSON文件字段：String转Array适配BigQuery

针对GCS中数百万JSON文件ing.info字段类型不一致（字符串/数组）的问题，提供三种高效处理方案，无需手动修改文件：

一、批量修正GCS源JSON文件（永久统一类型） #

适合希望从根源解决类型不一致的场景，利用GCP服务实现批量+自动处理：

1. 存量文件批量处理 #

用Python脚本结合GCS客户端库批量遍历并修改文件，核心逻辑是检测ing.info类型，将字符串转为单元素数组：

import json
from google.cloud import storage

def fix_ing_info(bucket_name, file_path):
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(file_path)
    
    # 读取并解析JSON
    content = json.loads(blob.download_as_text())
    if "ing" in content and isinstance(content["ing"]["info"], str):
        content["ing"]["info"] = [content["ing"]["info"]]
    
    # 覆盖原文件（建议先备份数据）
    blob.upload_from_string(json.dumps(content), content_type="application/json")

# 批量执行：遍历桶内所有JSON文件
def batch_fix(bucket_name):
    for blob in storage.Client().list_blobs(bucket_name):
        if blob.name.endswith(".json"):
            fix_ing_info(bucket_name, blob.name)

# 调用示例：batch_fix("your-gcs-bucket-name")

可借助GCP Batch服务或Cloud Run Jobs运行脚本，提升批量处理效率。

2. 新增文件自动处理 #

配置GCS触发器+Cloud Functions，新文件存入时自动修正类型：

• 创建Cloud Function，触发条件设为GCS桶的最终创建事件
• 函数逻辑与fix_ing_info一致，确保新文件存入时立即统一类型

二、BigQuery查询层转换（无需修改源文件） #

适合快速导入正式表、无需改动GCS文件的场景：

1. 手动指定外部表Schema #

创建外部表时，强制将ing.info定义为STRING类型（兼容两种格式），Schema示例：

[
  {
    "name": "ing",
    "type": "RECORD",
    "fields": [
      {"name": "info", "type": "STRING"},
      {"name": "details", "type": "ARRAY<STRING>"}
    ]
  }
]

2. 转换并导入正式表 #

写SQL将ing.info统一转为数组类型，直接插入正式表：

CREATE OR REPLACE TABLE `your-project.your-dataset.target_table` AS
SELECT
  -- 保留其他字段，替换ing字段为统一类型
  *,
  STRUCT(
    CASE
      -- 检测原字段是否为数组格式
      WHEN JSON_TYPE(ing.info) = 'ARRAY' THEN PARSE_JSON(ing.info)
      -- 字符串转为单元素数组
      ELSE [ing.info]
    END AS info,
    ing.details AS details
  ) AS ing
FROM `your-project.your-dataset.external_table`

如果外部表已自动推断为混合类型，可先删除外部表，用上述Schema重新创建。

三、Dataflow流式/批处理ETL（海量数据高效处理） #

针对持续流入的百万级数据，构建Dataflow管道实现实时转换：

import json
import apache_beam as beam

def normalize_ing_field(element):
    data = json.loads(element)
    if "ing" in data:
        info = data["ing"].get("info")
        if isinstance(info, str):
            data["ing"]["info"] = [info]
    return data

with beam.Pipeline() as p:
    (p
     # 读取GCS存量文件，或配置流式读取新文件
     | "Read JSON from GCS" >> beam.io.ReadFromText("gs://your-bucket/**/*.json")
     | "Normalize ing.info" >> beam.Map(normalize_ing_field)
     | "Write to BigQuery" >> beam.io.WriteToBigQuery(
         table="your-project.your-dataset.target_table",
         schema="ing:RECORD<info:ARRAY<STRING>, details:ARRAY<STRING>>",
         write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
     )
    )

该方案支持批处理存量数据+流式处理新增数据，适合超大规模数据场景。

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