如何通过Apache Airflow实现BigQuery到Spanner的每日批量加载与切换？

云上老周

2026-5-21

刚好我之前处理过类似的GCP大数据量迁移+Airflow编排的场景，来给你一步步拆解解决方案：

一、BigQuery到Spanner的基础批量加载方法

目前GCP生态里有两种主流的基础方案，适配不同场景：

GCS中转+Spanner原生导入
适合无数据转换需求的纯批量迁移：先把BigQuery数据导出到GCS（优先选Parquet格式，压缩率高、读写快），再用Spanner的原生导入能力加载数据。
示例gcloud命令：
```
gcloud spanner databases import --instance=你的实例ID --database=你的数据库ID gs://你的存储桶路径/data/*.parquet
```
注意：GCS存储桶要和Spanner实例在同一区域，避免跨区域传输的成本和延迟。
Dataflow直接读写
适合需要做数据清洗、字段转换的场景：用Dataflow的BigQuery读取连接器和Spanner写入连接器，直接从BigQuery拉取数据写入Spanner。可以用Python/Java编写管道逻辑，后续能直接用Airflow调度。

二、80GB/9亿行量级的加载优化

针对这个级别的数据量，要重点优化吞吐量和成本：

分区并行处理：把BigQuery表按主键分片导出到GCS的多个文件，然后并行启动Spanner导入任务（或用Dataflow的并行分区），最大化利用Spanner的分布式写入能力。
临时扩容Spanner节点：加载前临时把Spanner节点数从常规值（比如3节点）调高到10+节点，完成后再降回去——这能大幅提升写入速度，且临时扩容的成本远低于长时间等待加载的损耗。
用批量写入API：如果用代码实现，Spanner客户端库的BatchClient或write_batch方法可以批量提交写入请求，减少网络往返开销，每批建议控制在1000-5000行（根据单条数据大小调整）。

三、无表重命名下的在线数据切换方案

因为Spanner不支持表重命名，双表+原子视图切换是最稳妥的方案：

提前创建两张结构完全一致的表：daily_agg_current（对外提供查询的表）和daily_agg_staging（每次批量加载的临时表）。
每次加载任务都把数据写入daily_agg_staging，完成后做数据校验（比如行数匹配、主键唯一性检查）。
用原子SQL替换视图指向：
```
CREATE OR REPLACE VIEW daily_agg AS SELECT * FROM daily_agg_staging;
```
这个视图是对外暴露的查询入口，替换操作是原子性的，用户端不会感知到切换过程。
切换完成后，可以删除daily_agg_current的旧数据（或保留3-7天作为备份），然后把daily_agg_staging清空，等待下一次加载。

四、Airflow实现全流程自动化

结合你们已有的Airflow环境，可以把整个流程编排成DAG，核心步骤如下：

BigQuery导出到GCS
用BigQueryToGCSOperator实现分区导出：

from airflow.providers.google.cloud.transfers.bigquery_to_gcs import BigQueryToGCSOperator

bq_to_gcs = BigQueryToGCSOperator(
    task_id="bq_export_to_gcs",
    source_project_dataset_table="你的项目ID.数据集ID.每日聚合表",
    destination_cloud_storage_uris=["gs://你的存储桶/daily_agg/*.parquet"],
    export_format="PARQUET",
    compression="SNAPPY",
    print_header=False,
)

GCS数据导入Spanner临时表
可以用BashOperator调用gcloud命令，或者用GCSToSpannerOperator（如果你的Airflow版本支持）：

from airflow.operators.bash import BashOperator

gcs_to_spanner_staging = BashOperator(
    task_id="gcs_import_to_spanner_staging",
    bash_command="gcloud spanner databases import --instance=你的实例ID --database=你的数据库ID gs://你的存储桶/daily_agg/*.parquet --table=daily_agg_staging",
)

原子切换视图
用SpannerExecuteSqlOperator执行视图替换SQL：

from airflow.providers.google.cloud.operators.spanner import SpannerExecuteSqlOperator

switch_agg_view = SpannerExecuteSqlOperator(
    task_id="switch_daily_agg_view",
    instance_id="你的实例ID",
    database_id="你的数据库ID",
    sql="CREATE OR REPLACE VIEW daily_agg AS SELECT * FROM daily_agg_staging;",
)

可选：清理旧数据/调整Spanner节点
可以用SpannerUpdateInstanceOperator临时扩容/缩容节点，或者用SpannerExecuteSqlOperator删除旧表数据。

五、编程方式实现批量加载

如果需要更灵活的自定义逻辑，也可以用GCP客户端库编写脚本：

from google.cloud import spanner, bigquery

# 初始化客户端
spanner_client = spanner.Client()
bq_client = bigquery.Client()

# 从BigQuery分批拉取数据
query_job = bq_client.query("SELECT * FROM 你的聚合表")
rows_iter = query_job.result(page_size=1000)

# 批量写入Spanner临时表
instance = spanner_client.instance("你的实例ID")
database = instance.database("你的数据库ID")

for page in rows_iter.pages:
    rows = list(page)
    with database.batch_update() as batch:
        batch.insert(
            table="daily_agg_staging",
            columns=["主键字段", "字段1", "字段2", ...],
            values=rows,
        )

不过对于9亿行的量级，更推荐GCS中转+Spanner原生导入的方案，速度会比直接拉取写入快很多。

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