无需ETL工具：GCS至BigQuery的非结构化JSON数据加载与转换

核心思路 #

用Python轻量脚本流式处理GCS中的JSON文件（避免全量加载占用内存），实现自定义转换逻辑后写入GCS，再通过BigQuery原生加载功能导入数据，全程无需专用ETL工具，低算力资源即可完成。

步骤1：编写Python流式转换脚本 #

通过google-cloud-storage库直接操作GCS文件，逐行/逐元素处理非结构化JSON，自定义转换逻辑后流式写入结构化JSON文件，全程内存占用极低。

依赖安装 #

pip install google-cloud-storage google-cloud-bigquery ijson

示例脚本 #

from google.cloud import storage
import json
import ijson

def transform_raw_json(raw_data):
    # 自定义转换逻辑：统一字段名、提取关键数据、处理缺失值
    structured_data = {
        "user_id": raw_data.get("user_id") or raw_data.get("id"),
        "event_time": raw_data.get("created_at") or raw_data.get("timestamp"),
        "event_content": raw_data.get("payload", {}).get("content"),
        "source_system": raw_data.get("source")
    }
    # 过滤无效数据（必填字段缺失则跳过）
    if structured_data["user_id"] and structured_data["event_time"]:
        return structured_data
    return None

def process_gcs_json(bucket_name, source_path, dest_path):
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    source_blob = bucket.blob(source_path)
    
    # 流式读取源文件，处理后流式写入目标文件
    with source_blob.open("rb") as source_file:
        dest_blob = bucket.blob(dest_path)
        with dest_blob.open("w") as dest_file:
            # 兼容两种JSON格式：换行分隔的JSON对象 或 大型JSON数组
            try:
                # 先尝试按行读取（换行分隔格式）
                for line in source_file:
                    raw_json = json.loads(line)
                    transformed = transform_raw_json(raw_json)
                    if transformed:
                        dest_file.write(json.dumps(transformed) + "\n")
            except json.JSONDecodeError:
                # 若为JSON数组，用ijson逐元素解析
                source_file.seek(0)
                for item in ijson.items(source_file, "item"):
                    transformed = transform_raw_json(item)
                    if transformed:
                        dest_file.write(json.dumps(transformed) + "\n")

# 配置GCS路径参数
BUCKET_NAME = "your-gcs-bucket"
RAW_JSON_PATH = "raw_data/unstructured_200gb.json"
CLEANED_JSON_PATH = "structured_data/cleaned_data.json"

process_gcs_json(BUCKET_NAME, RAW_JSON_PATH, CLEANED_JSON_PATH)

步骤2：拆分大文件优化处理效率 #

200GB单文件处理较慢，可先用gsutil拆分文件为多个小文件，并行处理提升速度：

# 将大文件按10万行拆分，生成多个小文件
gsutil cat gs://your-gcs-bucket/raw_data/unstructured_200gb.json | split -l 100000 - gs://your-gcs-bucket/raw_data/split_

之后修改脚本循环处理所有拆分后的小文件即可。

步骤3：加载结构化JSON到BigQuery #

用BigQuery原生命令行工具bq完成加载，无需额外算力：

方式1：指定Schema加载（推荐） #

先编写schema.json定义表结构：

[
    {"name": "user_id", "type": "STRING", "mode": "REQUIRED"},
    {"name": "event_time", "type": "TIMESTAMP", "mode": "REQUIRED"},
    {"name": "event_content", "type": "STRING", "mode": "NULLABLE"},
    {"name": "source_system", "type": "STRING", "mode": "NULLABLE"}
]

执行加载命令：

bq load \
    --source_format=NEWLINE_DELIMITED_JSON \
    --schema=schema.json \
    your-project.your_dataset.target_table \
    gs://your-gcs-bucket/structured_data/*.json

方式2：自动检测Schema（快速验证） #

bq load \
    --source_format=NEWLINE_DELIMITED_JSON \
    --autodetect \
    your-project.your_dataset.target_table \
    gs://your-gcs-bucket/structured_data/*.json

低算力适配要点 #

• 用普通云VM（如n1-standard-1）或本地机器运行脚本，流式处理无需大内存（单进程内存占用通常<1GB）
• 拆分文件后可多进程并行处理，避免单进程瓶颈
• 处理过程中添加日志记录，方便排查无效JSON行或转换错误

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