如何通过Python API高效批量更新BigQuery万行以上数据？

1. 用BigQuery DML批量更新（最简便高效） #

直接通过Python客户端执行批量DML语句，这比逐行或异步逐行快得多——BigQuery会在服务器端完成批量处理，避免频繁的网络请求开销。

如果是针对特定条件批量更新：

from google.cloud import bigquery

client = bigquery.Client()

# 构造批量更新语句
update_query = """
UPDATE `你的项目ID.你的数据集.目标表`
SET 字段1 = 新值1, 字段2 = 新值2
WHERE 筛选字段 = '目标条件'
"""

# 执行并等待完成
query_job = client.query(update_query)
query_job.result()

如果更新数据来自本地数据集（比如Pandas DataFrame），可以先把数据写入临时表，再通过JOIN实现批量更新：

import pandas as pd

# 假设df是包含更新数据的DataFrame，需包含主键用于匹配
temp_table_ref = client.dataset("你的数据集").table("临时更新表")
# 写入临时表
client.load_table_from_dataframe(df, temp_table_ref).result()

# 通过JOIN批量更新主表
update_query = """
UPDATE `你的项目ID.你的数据集.目标表` t
SET t.字段1 = u.新值1, t.字段2 = u.新值2
FROM `你的项目ID.你的数据集.临时更新表` u
WHERE t.主键字段 = u.主键字段
"""

client.query(update_query).result()
# 清理临时表
client.delete_table(temp_table_ref)

2. 用MERGE实现批量UPSERT（插入/更新二合一） #

如果需要实现“存在则更新，不存在则插入”的逻辑，用MERGE语句搭配临时表是最优解，原子性强且效率极高：

job_config = bigquery.LoadJobConfig(
    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    create_disposition=bigquery.CreateDisposition.CREATE_IF_NEEDED
)

# 写入待更新数据到临时表
client.load_table_from_dataframe(df, temp_table_ref, job_config=job_config).result()

# 执行MERGE
merge_query = """
MERGE `你的项目ID.你的数据集.目标表` t
USING `你的项目ID.你的数据集.临时更新表` u
ON t.主键字段 = u.主键字段
WHEN MATCHED THEN
  UPDATE SET 字段1 = u.字段1, 字段2 = u.字段2
WHEN NOT MATCHED THEN
  INSERT (主键字段, 字段1, 字段2) VALUES (u.主键字段, u.字段1, u.字段2)
"""

client.query(merge_query).result()
client.delete_table(temp_table_ref)

为啥不推荐asyncio逐行更新？ #

asyncio异步逐行本质还是多次网络请求，虽然比同步逐行快，但远不如服务器端批量处理高效，还要额外维护异步逻辑，复杂度更高。上面的DML/MERGE方案只用几行代码就能搞定，性能提升几个量级，维护成本还低。

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