在 Python 中使用 BigQuery 通常是通过 Google 官方提供的客户端库 google-cloud-bigquery 来实现的，以下是使用 Python 连接 BigQuery 并执行常见操作的完整指南：
✅ 1. 安装依赖
pip install google-cloud-bigquery
✅ 2. 认证方式
BigQuery 使用 Google Cloud 认证,推荐方式：
使用服务账号密钥文件（推荐用于生产环境）
- 在 Google Cloud Console 中创建一个服务账号。
- 下载 JSON 密钥文件。
- 设置环境变量：
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/your-service-account-file.json"
使用本地 gcloud CLI 认证（适合开发环境）
gcloud auth application-default login
✅ 3. 基本使用示例
1 初始化客户端
from google.cloud import bigquery # 自动从环境变量或默认凭据加载认证信息 client = bigquery.Client()
2 列出所有数据集
datasets = list(client.list_datasets())
print("Datasets in project:", [ds.dataset_id for ds in datasets])
3 查询数据
query = """ SELECT name, COUNT() as count FROM `bigquery-public-data.usa_names.usa_1910_current` WHERE state = 'TX' GROUP BY name ORDER BY count DESC LIMIT 10 """ # 执行查询 query_job = client.query(query) results = query_job.result() # 等待查询完成 # 打印结果 for row in results: print(f"{row.name}: {row.count}")
4 将查询结果保存为 DataFrame（可选）
import pandas as pd df = query_job.to_dataframe() print(df.head())
⚠️ 需要安装
pandas和google-cloud-bigquery[pandas]
5 创建表并加载数据
from google.cloud import bigquery
client = bigquery.Client()
# 定义表 ID
table_id = "your-project.your_dataset.your_table"
# 创建表配置
schema = [
bigquery.SchemaField("name", "STRING"),
bigquery.SchemaField("age", "INTEGER"),
]
table = bigquery.Table(table_id, schema=schema)
table = client.create_table(table) # 创建表
print(f"Created table {table.project}.{table.dataset_id}.{table.table_id}")
6 插入数据
rows = [
("Alice", 30),
("Bob", 25),
("Charlie", 35),
]
errors = client.insert_rows_json(table_id, rows)
if errors == []:
print("Successfully inserted rows.")
else:
print("Errors:", errors)
7 删除表
client.delete_table(table_id, delete_contents=True)
print(f"Deleted table {table_id}")
✅ 4. 高级用法
1 使用标准 SQL 参数化查询
query = """
SELECT
FROM `your-project.your_dataset.your_table`
WHERE age > @min_age
"""
query_params = [
bigquery.ScalarQueryParameter("min_age", "INTEGER", 25)
]
job_config = bigquery.QueryJobConfig(query_parameters=query_params)
query_job = client.query(query, job_config=job_config)
results = query_job.result()
for row in results:
print(row)
2 分页查询
query_job = client.query("SELECT FROM `your-project.your_dataset.your_table`")
results = query_job.result(page_size=100) # 每页100条
for page in results.pages:
for row in page:
print(row)
3 异步查询
from google.cloud import bigquery
client = bigquery.Client()
query = "SELECT FROM `bigquery-public-data.usa_names.usa_1910_current` LIMIT 100"
query_job = client.query(query)
# 异步获取结果
results = query_job.result() # 阻塞直到完成
for row in results:
print(row)
✅ 5. 最佳实践
| 实践 | 说明 |
|---|---|
| 使用参数化查询 | 防止 SQL 注入，提高性能 |
| 限制查询范围 | 使用 LIMIT 或分区表减少扫描数据量 |
| 使用分区/聚类表 | 提高查询效率，降低成本 |
| 监控查询成本 | 使用 query_job.total_bytes_processed 监控数据扫描量 |
| 使用连接池/复用客户端 | 避免频繁创建客户端实例 |
✅ 6. 常见问题
Q: 认证失败？
- 确保
GOOGLE_APPLICATION_CREDENTIALS环境变量指向正确的 JSON 文件。 - 或使用
gcloud auth application-default login。
Q: 查询超时？
- 增加
job_config.timeout_ms或检查查询是否扫描了大量数据。
Q: 权限不足？
- 确保服务账号具有
BigQuery Data Viewer或更高权限。
首发原创文章，作者：王坚，如若转载，请注明出处：https://test.idctop.com/article/476569.html
