使用JSON文件创建Google BigQuery表失败，求批量转NDJSON方法

方法1：Python脚本批量转换 #

• 适用于本地或能运行Python的环境，处理任意大小的JSON文件
• 示例代码（针对数组格式的原始JSON）：

import json

# 替换为你的输入输出文件路径
input_file = "GCP_Cloud_Storage_HR_Analytics_POC-Changed_data_14-07[1].json"
output_file = "converted_hr_data.ndjson"

with open(input_file, 'r', encoding='utf-8') as f_in, open(output_file, 'w', encoding='utf-8') as f_out:
    # 加载整个JSON数组
    data = json.load(f_in)
    # 遍历每个对象，逐行写入新文件
    for item in data:
        json.dump(item, f_out)
        f_out.write('\n')

• 特殊情况处理：如果原始JSON是嵌套结构（比如根节点是{"records": [ ... ]}），把data = json.load(f_in)改成data = json.load(f_in)["records"]即可，根据实际结构调整。

方法2：jq命令行工具快速转换 #

• 适合熟悉命令行的用户，GCP Cloud Shell默认自带jq，无需额外安装
• 执行命令（针对数组格式JSON）：

jq -c '.[]' GCP_Cloud_Storage_HR_Analytics_POC-Changed_data_14-07[1].json > converted_hr_data.ndjson

• 参数说明：
  • -c：输出紧凑的单行JSON
  • .[]：遍历JSON数组的每个元素，逐个输出
• 嵌套结构适配：如果原始JSON根节点是{"data": [ ... ]}，命令改为：

jq -c '.data[]' input.json > output.ndjson

方法3：GCP云端直接转换（无需下载文件） #

• 用Cloud Functions自动处理Cloud Storage中的JSON文件：
  1. 创建Cloud Function，触发方式选择Cloud Storage（当文件上传时触发）
  2. 核心代码（Python）：

import json
from google.cloud import storage

def convert_json_to_ndjson(event, context):
    bucket_name = event['bucket']
    file_name = event['name']
    # 跳过已转换的NDJSON文件
    if file_name.endswith('.ndjson'):
        return
    
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    
    # 读取原始JSON内容
    content = blob.download_as_text()
    data = json.loads(content)
    
    # 生成NDJSON内容
    ndjson_content = '\n'.join([json.dumps(item) for item in data])
    
    # 上传转换后的文件到指定目录
    output_blob = bucket.blob(f"converted/{file_name.replace('.json', '.ndjson')}")
    output_blob.upload_from_string(ndjson_content, content_type='application/x-ndjson')

3. 配置权限：确保函数拥有Cloud Storage的读写权限

转换结果验证 #

转换完成后可以做两步检查：

• 打开NDJSON文件，确认每行是一个独立的JSON对象
• 在BigQuery中选择转换后的文件创建表，确认解析错误消失

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