Amazon S3 JSON文件转NDJSON导入BigQuery问题咨询

核心处理逻辑 #

你需要将外层公共元字段edwLoadDate、edwRecordSource注入到data数组的每一条子记录中，再将每条处理后的子记录序列化为单行JSON，最终生成的每行独立的文件即为BigQuery支持的NDJSON格式。 #

方案1：jq命令快速处理（适合小体积JSON文件） #

无需开发复杂代码，安装jq工具后执行以下命令即可完成转换：

jq -c '. as $root | .data[] | . + {edwLoadDate: $root.edwLoadDate, edwRecordSource: $root.edwRecordSource}' 源文件.json > 输出文件.ndjson

处理后的NDJSON示例输出如下：

{"data":{"lastName":"John","firstName":"Doe","email":"john@example123.com"},"edwSequence":"0","edwLoadDate":"2017-01-18T18-08-44","edwRecordSource":"MAILCHIMP"}
{"data":{"lastName":"John","firstName":"Doe","email":"john@example123.com"},"edwSequence":"1","edwLoadDate":"2017-01-18T18-08-44","edwRecordSource":"MAILCHIMP"}

方案2：Python脚本处理（适合大文件/批量处理场景） #

如果需要处理S3上的批量文件或者大体积JSON，可使用以下Python脚本，支持直接读写S3文件：

import json
import boto3

s3 = boto3.client('s3')

def convert_json_to_ndjson(bucket, source_key, target_key):
    # 读取S3源JSON文件
    resp = s3.get_object(Bucket=bucket, Key=source_key)
    raw_data = json.loads(resp['Body'].read().decode('utf-8'))
    
    # 提取外层公共元字段
    common_meta = {
        'edwLoadDate': raw_data['edwLoadDate'],
        'edwRecordSource': raw_data['edwRecordSource']
    }
    
    # 生成NDJSON内容
    ndjson_lines = []
    for item in raw_data['data']:
        # 合并公共字段与单条记录
        merged_item = {**item, **common_meta}
        ndjson_lines.append(json.dumps(merged_item, ensure_ascii=False))
    
    # 写入S3目标路径
    s3.put_object(
        Bucket=bucket,
        Key=target_key,
        Body='\n'.join(ndjson_lines).encode('utf-8')
    )

# 调用示例
convert_json_to_ndjson('你的S3桶名', '源JSON路径/xxx.json', '输出NDJSON路径/xxx.ndjson')

如果是GB级以上超大文件，可替换为ijson流式解析库避免内存溢出。 #

方案3：Lambda自动化流程（生产环境批量作业） #

• 配置S3桶的对象创建事件，触发Lambda函数自动处理新上传的源JSON文件
• Lambda执行转换逻辑后将NDJSON写入指定S3路径
• 配置BigQuery定时加载任务，直接从该路径读取NDJSON完成自动导入

注意事项 #

• 生成NDJSON时需确保每行是完整独立的JSON结构，无多余逗号、换行，避免BigQuery导入报错
• 正式全量转换前可先用小样本测试BigQuery导入，确认表结构字段匹配无误
• 若转换后的字段层级不符合预期，可调整合并逻辑，将内层data的字段直接展开到根层级

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