如何通过Airflow DAG将BigQuery视图导出为GCS存储桶中的CSV文件

针对你遇到的BigQueryToCloudStorageOperator无法导出视图的问题，在Airflow 1.10.12版本中有两种可行的解决思路：

方法一：通过临时中转表导出 #

核心思路是先将视图数据写入临时表，再用常规算子导出临时表：

1. 创建临时表：用BigQueryOperator执行查询，将视图结果写入临时表（支持会话级临时表或常规表）
2. 导出临时表：用BigQueryToCloudStorageOperator将临时表数据导出到GCS
3. 清理临时表（可选）：如果使用常规临时表，导出完成后可删除以节省存储资源

示例代码：

from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
from airflow.contrib.operators.bigquery_to_gcs import BigQueryToCloudStorageOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2023, 1, 1),
}

with DAG('export_bq_view_to_gcs', default_args=default_args, schedule_interval='@daily') as dag:
    # 1. 将视图数据写入临时表
    create_temp_table = BigQueryOperator(
        task_id='create_temp_table',
        sql="""
            CREATE OR REPLACE TABLE `your-project.your-dataset.temp_export_table`
            AS SELECT * FROM `your-project.your-dataset.my_view`
        """,
        use_legacy_sql=False,
        bigquery_conn_id='google_cloud_default'
    )

    # 2. 导出临时表到GCS CSV
    export_to_gcs = BigQueryToCloudStorageOperator(
        task_id='export_to_gcs',
        source_project_dataset_table='your-project.your-dataset.temp_export_table',
        destination_cloud_storage_uris=['gs://your-bucket/path/to/export-*.csv'],
        export_format='CSV',
        field_delimiter=',',
        print_header=True,
        bigquery_conn_id='google_cloud_default'
    )

    # 3. 可选：删除临时表
    delete_temp_table = BigQueryOperator(
        task_id='delete_temp_table',
        sql="DROP TABLE IF EXISTS `your-project.your-dataset.temp_export_table`",
        use_legacy_sql=False,
        bigquery_conn_id='google_cloud_default'
    )

    create_temp_table >> export_to_gcs >> delete_temp_table

方法二：用PythonOperator结合BigQueryHook直接导出 #

无需创建临时表，直接通过BigQueryHook提交导出任务，将视图查询结果直接导出到GCS：

示例代码：

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.contrib.hooks.bigquery_hook import BigQueryHook
from datetime import datetime

def export_view_to_gcs():
    hook = BigQueryHook(bigquery_conn_id='google_cloud_default', use_legacy_sql=False)
    client = hook.get_client()

    # 构造导出配置
    job_config = client.job_config()
    job_config.destination_format = 'CSV'
    job_config.field_delimiter = ','
    job_config.print_header = True

    # 定义视图查询语句
    query = "SELECT * FROM `your-project.your-dataset.my_view`"

    # 提交导出作业
    extract_job = client.extract_table(
        query,
        'gs://your-bucket/path/to/export-*.csv',
        job_config=job_config
    )

    # 等待作业完成
    extract_job.result()

default_args = {
    'start_date': datetime(2023, 1, 1),
}

with DAG('export_bq_view_to_gcs_direct', default_args=default_args, schedule_interval='@daily') as dag:
    export_task = PythonOperator(
        task_id='export_view_to_gcs',
        python_callable=export_view_to_gcs
    )

两种方法对比 #

• 方法一：逻辑简单，无需额外代码开发，适合小到中等数据量；缺点是需要额外存储临时表数据，会产生少量BigQuery存储成本。
• 方法二：无需临时表，数据直接导出，更高效；缺点是需要自行编写Python逻辑处理作业提交和等待。

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