如何高效将超大规模BigQuery表的单列数据加载到Python列表中

方案1：使用BigQuery存储API直接拉取结果（首选） #

默认的query()方法返回的行迭代器是逐页拉取数据，吞吐量极低，而且会在内存中缓存所有行，很容易触发OOM。谷歌官方提供的BigQuery存储API专门用于大批量数据读取，吞吐量可以提升10倍以上，内存占用也更低。
代码示例：

from google.cloud import bigquery
from google.cloud import bigquery_storage

bq_client = bigquery.Client()
bq_storage_client = bigquery_storage.BigQueryReadClient()

query = """ SELECT column FROM table GROUP BY column """
# 运行查询得到作业结果
query_job = bq_client.query(query)
# 用存储API读取结果转DataFrame，效率远高于逐行迭代
rows = query_job.result().to_dataframe(
    bqstorage_client=bq_storage_client
)
# 直接转列表
column_list = rows["column"].tolist()

需要先安装依赖：pip install google-cloud-bigquery-storage pyarrow

方案2：导出到GCS后本地读取（适合去重后结果仍超千万级的场景） #

如果去重后的字段值量级还是极高，建议先把查询结果导出为GCS上的Parquet文件，再下载到本地用pyarrow读取，内存占用会进一步降低：

1. 运行查询+导出作业

query = """ SELECT column FROM table GROUP BY column """
# 配置查询结果存入临时表
job_config = bigquery.QueryJobConfig(
    destination_format = bigquery.DestinationFormat.PARQUET,
    write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE,
    destination = "你的项目ID.你的数据集名.临时表名"
)
query_job = bq_client.query(query, job_config=job_config)
query_job.result()

# 导出临时表到GCS路径
extract_job = bq_client.extract_table(
    "你的项目ID.你的数据集名.临时表名",
    "gs://你的存储桶路径/export_*.parquet",
    destination_format = bigquery.DestinationFormat.PARQUET
)
extract_job.result()

2. 下载Parquet文件后读取转列表：

import pyarrow.parquet as pq

table = pq.read_table("本地Parquet文件路径", columns=["column"])
column_list = table["column"].to_pylist()

其他优化建议 #

• 如果该字段的去重值不需要实时更新，可以提前预计算去重结果存在小表里，每次直接读小表即可，不需要每次都运行GROUP BY逻辑
• 如果业务逻辑不需要一次性把所有值加载到内存，可以用分页迭代的方式分批处理，完全避免OOM问题：

page_size = 10000
query_result = bq_client.query(query).result(page_size=page_size)
for page in query_result.pages:
    page_list = [row.column for row in page]
    # 分批处理当前页列表，不需要整合成一个大列表

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