使用Airflow运行BigQuery查询直接导出Parquet数据到GCS的方案咨询

完全可以跳过中间表写入步骤，直接通过BigQuery原生的EXPORT DATA标准SQL语法实现「带过滤条件的查询结果直接导出到GCS」，仅需要单个BigQuery算子即可完成全流程，不需要拆分为两个任务。

实现代码示例 #

你可以直接替换原有的两个任务为如下单个任务：

direct_export = bigquery_operator.BigQueryOperator(
    task_id='direct_export_bq_to_gcs',
    sql="""
    EXPORT DATA
    OPTIONS(
        uri = '{output_uri}',
        format = 'PARQUET',
        compression = 'SNAPPY' -- 可选配置，不需要压缩可以删除该参数
    )
    AS
    -- 这里直接写你原来的带过滤条件的SELECT查询即可
    <select query with filters>
    """.format(
        output_uri='gs://<你的GCS存储桶路径>/导出文件名前缀-*.parquet',
        date=date1
    ),
    use_legacy_sql=False,
    location="southamerica-east1"
)

核心说明与注意事项 #

• 该方案底层是BigQuery直接将查询结果流式导出到GCS，不会在你的项目中生成任何中间表，既节省存储资源，也省去了中间表的生命周期管理成本
• GCS导出路径末尾必须加*通配符：如果查询结果量超过1GB，BigQuery会自动按照分片规则导出多个文件，避免单文件过大无法处理
• 你可以根据业务需要调整PARQUET的压缩配置，支持SNAPPY、GZIP等压缩格式，不需要压缩直接删除compression参数即可
• 如果你使用的是Airflow 2.x及以上版本，BigQueryOperator已迁移到apache-airflow-providers-google包的BigQueryInsertJobOperator，除导入路径外写法完全一致，无需调整核心逻辑

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