如何用Python分批读取BigQuery大表数据并逐批完成预处理

以下是3种生产环境常用的可行实现，默认单批次读取10000行，可根据内存情况调整批次大小：

方案1：官方google-cloud-bigquery客户端原生分页（推荐） #

这是Google官方推荐的读取方式，客户端底层自动处理分页令牌，无重复、漏读问题，性能最优。

• 依赖安装：pip install google-cloud-bigquery
• 示例代码：

from google.cloud import bigquery

# 初始化BigQuery客户端
client = bigquery.Client(project="替换为你的GCP项目ID")
# 指定目标表
table_ref = client.dataset("替换为数据集名", project="替换为你的GCP项目ID").table("替换为表名")

# 配置单批次读取行数
row_iterator = client.list_rows(table_ref, page_size=10000)

# 逐批处理数据
for page in row_iterator.pages:
    # 可根据需要转为DataFrame或直接处理Row对象
    batch_df = page.to_dataframe()
    # 此处写入你的预处理逻辑
    print(f"当前批次处理行数：{len(batch_df)}")
    # 处理完成后手动释放内存
    del batch_df

方案2：pandas read_gbq分块读取 #

适合习惯用pandas做数据处理的场景，代码极简，直接返回DataFrame格式的批次数据。

• 依赖安装：pip install pandas-gbq google-cloud-bigquery
• 示例代码：

import pandas as pd

# 建议仅查询需要的字段，不要用SELECT * 减少数据传输量
query_sql = "SELECT * FROM `替换为你的项目ID.替换为数据集名.替换为表名`"

# 配置chunksize返回分块迭代器
chunk_iter = pd.read_gbq(
    query=query_sql,
    project_id="替换为你的GCP项目ID",
    chunksize=10000,
    dialect="standard"
)

# 逐批处理
for batch_df in chunk_iter:
    # 预处理逻辑
    print(f"当前批次行数：{len(batch_df)}")
    del batch_df

方案3：手动LIMIT+OFFSET分页 #

适合需要自定义分页逻辑、从指定偏移位置开始读取的场景。

• 注意：偏移量超过100万行后查询性能会明显下降，超大表不推荐该方案。
• 示例代码：

from google.cloud import bigquery

client = bigquery.Client(project="替换为你的GCP项目ID")
BATCH_SIZE = 10000
offset = 0
dataset_name = "替换为数据集名"
table_name = "替换为表名"

while True:
    sql = f"""
        SELECT * FROM `{client.project}.{dataset_name}.{table_name}`
        LIMIT {BATCH_SIZE} OFFSET {offset}
    """
    batch_df = client.query(sql).to_dataframe()
    # 读取为空时结束循环
    if len(batch_df) == 0:
        break
    # 预处理逻辑
    print(f"处理偏移量{offset}~{offset+BATCH_SIZE}行，共{len(batch_df)}行")
    offset += BATCH_SIZE
    del batch_df

通用优化建议 #

• 所有方案都建议仅查询需要的字段，避免读取无效列浪费内存和带宽
• 预处理完成后及时删除批次变量，可配合import gc; gc.collect()手动触发垃圾回收加速内存释放
• 能在BigQuery中通过SQL完成的过滤、聚合预处理，优先在SQL层完成，减少需要拉取到本地的数据量

内容的提问来源于stack exchange，提问作者Rafał Manda