Airflow实现BigQuery查询结果直接导出CSV并邮件发送方案咨询

针对不需要临时表、不需要GCS中间存储、避免大体积数据走XCom的需求，以下两种方案均可直接落地：

方案1：Python流式拉取结果写入本地压缩CSV（内存占用极低） #

该方案利用BigQuery查询后服务端自动生成的临时结果集（无需用户手动创建临时表，结果默认保留24小时），逐批次拉取数据写入本地，全程无中间存储，内存仅保留当前批次数据：

• 第一个任务仅执行查询并返回Job ID，仅需要传递Job ID到XCom，数据量极小完全符合XCom使用规范
• 第二个任务接收Job ID后，分页拉取查询结果，流式写入压缩CSV文件

示例代码如下：

from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook
import csv
import gzip

# 执行BQ查询仅返回Job ID的任务函数
def run_bq_query(**context):
    bq_hook = BigQueryHook(gcp_conn_id="你的GCP连接ID", use_legacy_sql=False)
    job = bq_hook.run_query(
        sql="替换为你的业务查询SQL",
        location="替换为你的BigQuery数据集所在区域，如US、asia-southeast1"
    )
    context["ti"].xcom_push(key="bq_job_id", value=job.job_id)

# 流式拉取结果写入压缩CSV的任务函数
def stream_to_compressed_csv(**context):
    bq_hook = BigQueryHook(gcp_conn_id="你的GCP连接ID", use_legacy_sql=False)
    job_id = context["ti"].xcom_pull(key="bq_job_id", task_ids="执行BQ查询的任务ID")
    location = "替换为你的BigQuery数据集所在区域"
    # 按页拉取结果，page_size可根据数据量调整
    results = bq_hook.get_job(job_id=job_id, location=location).result(page_size=1000)
    
    # 直接写入gzip压缩文件，无需存储未压缩版本
    with gzip.open("/本地存储路径/日报表.csv.gz", "wt", encoding="utf-8") as f:
        writer = csv.writer(f)
        # 写入表头
        writer.writerow([field.name for field in results.schema])
        # 逐行写入数据
        for row in results:
            writer.writerow(row.values())

该方案适配千万级以下数据量场景，全程不需要创建任何用户侧临时表、不需要上传GCS，内存占用稳定。

方案2：调用bq命令行工具一键导出（代码量最少） #

如果Airflow Worker节点已预安装Google Cloud SDK，可直接通过BashOperator调用官方bq工具完成导出，无需自行处理分页、类型转换逻辑：

bq query --use_legacy_sql=false --format=csv \
"替换为你的业务查询SQL" \
| gzip > /本地存储路径/日报表.csv.gz

该方案适合百万级以下数据量场景，逻辑简单、稳定性高。

后续操作 #

拿到压缩后的CSV文件后，直接使用Airflow自带的EmailOperator将文件作为附件发送即可，可在邮件任务完成后追加清理本地文件的步骤，避免磁盘占用过高。

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