如何批量将大CSV文件通过Pandas写入BigQuery以避免内存错误

针对单文件达数GB的场景，以下几种方案可以避免内存错误，实现高效批量导入：

方案1：Pandas分块读取+to_gbq追加写入 #

利用Pandas的分块读取功能，将大文件拆分成小批次处理，逐批写入BigQuery：

import pandas as pd

# 配置目标表信息
PROJECT_ID = "your-project-id"
DATASET_ID = "your-dataset-id"
TABLE_ID = f"{PROJECT_ID}.{DATASET_ID}.target-table"

# 分块读取CSV（chunksize根据内存调整，建议5-20万行）
for chunk in pd.read_csv("large_file.csv", chunksize=100000):
    # 可选：数据清洗（如处理缺失值、字段格式转换）
    chunk = chunk.dropna(subset=["critical_column"])
    # 逐批追加写入BigQuery
    chunk.to_gbq(
        destination_table=TABLE_ID,
        project_id=PROJECT_ID,
        if_exists="append",
        chunksize=50000  # 内部再细分批次，降低单请求压力
    )

• 注意：chunksize需根据本地内存容量调整，避免单块数据占用过多内存；确保CSV字段与BigQuery表schema匹配，不匹配时可通过schema参数手动指定。

方案2：BigQuery官方客户端库批量插入 #

使用google-cloud-bigquery的原生批量插入接口，直接按行分批次上传，无需加载整个文件到内存：

from google.cloud import bigquery
import csv

client = bigquery.Client(project="your-project-id")
TABLE_ID = "your-project-id.your-dataset.target-table"

BATCH_SIZE = 10000  # 每批次上传行数
rows = []

with open("large_file.csv", "r", encoding="utf-8") as f:
    reader = csv.reader(f)
    next(reader)  # 跳过表头
    for row in reader:
        rows.append(row)
        if len(rows) >= BATCH_SIZE:
            # 批量插入数据
            errors = client.insert_rows_json(TABLE_ID, rows)
            if errors:
                print(f"批量插入失败：{errors}")
            rows = []
    # 处理剩余未上传的行
    if rows:
        errors = client.insert_rows_json(TABLE_ID, rows)
        if errors:
            print(f"剩余行插入失败：{errors}")

• 注意：需提前配置BigQuery认证（如设置GOOGLE_APPLICATION_CREDENTIALS环境变量）；若表头与表schema字段顺序不一致，需手动映射字段顺序。

方案3：GCS中转+BigQuery批量加载（超大型文件首选） #

对于10GB以上的超大文件，先上传到Google Cloud Storage（GCS），再通过BigQuery的批量加载API导入，完全规避本地内存限制：

from google.cloud import bigquery
from google.cloud import storage

# 1. 上传CSV到GCS
storage_client = storage.Client(project="your-project-id")
bucket = storage_client.get_bucket("your-gcs-bucket")
blob = bucket.blob("data/large_file.csv")
blob.upload_from_filename("large_file.csv")

# 2. 从GCS加载到BigQuery
bq_client = bigquery.Client(project="your-project-id")
TABLE_ID = "your-project-id.your-dataset.target-table"

job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.CSV,
    skip_leading_rows=1,
    autodetect=True,  # 自动检测schema，也可手动指定
    write_disposition=bigquery.WriteDisposition.WRITE_APPEND
)

gcs_uri = f"gs://your-gcs-bucket/data/large_file.csv"
load_job = bq_client.load_table_from_uri(gcs_uri, TABLE_ID, job_config=job_config)

load_job.result()  # 等待加载任务完成
print(f"加载完成，共写入{load_job.output_rows}条数据")

• 注意：GCS桶与BigQuery建议在同一区域，减少跨区域传输成本；该方式由BigQuery自动处理分片和并行加载，效率最高。

内容的提问来源于stack exchange，提问作者Zabir Al Nazi Nabil