如何将 BigQuery 数据导出到 GCS？

Posted 2023-03-24

【中文标题】如何将 BigQuery 数据导出到 GCS？

【英文标题】：How to export BigQuery data to GCS?

【发布时间】：2021-11-15 15:34:46

【问题描述】：

我是一名 Airflow 新手，正在尝试从 BigQuery 收集数据并将查询到的数据导出到 Google Cloud Storage。我创建了一个任务来检查查询的数据是否存在：

t1 = BigQueryCheckOperator(
    task_id="bq_check_covid_data_exists",
    sql="""
        SELECT COUNT(*) > 0
        FROM bigquery-public-data.covid19_italy.data_by_region
        WHERE DATE(date) = DATE_ADD(DATE " ds ", INTERVAL -2 DAY)
    """,
    use_legacy_sql=False,
    dag=dag
)

现在，我想查询这些数据，然后导出到 GCS，可能使用BigQueryToCloudStorageOperator idk。我需要使用哪些操作员来完成这种工作？如果您提供一些代码示例或来源，我会很高兴。

【参考方案1】：

你可以把它分成两步：

BigQuery 运算符使用您的数据创建临时表 用于存储在云存储上的 BQtoGCS 运算符

第一个任务可以这样解决：

BigQueryOperator(
        task_id="execute_check",
        bigquery_conn_id=google_cloud_connection_id,
        bql=query,
        destination_dataset_table="project:tmp_dataset.table_name",
        create_disposition="CREATE_IF_NEEDED",
        write_disposition="WRITE_TRUNCATE",
        use_legacy_sql=False,
        allow_large_results=True,
        dag=dag
    )

然后在 GCS 上上传：

BigQueryToCloudStorageOperator(
        task_id="save_data_on_storage",
        source_project_dataset_table="project:tmp_dataset.table_name",
        destination_cloud_storage_uris=["gs://bucket/path/file.csv"],
        export_format="CSV",
        field_delimiter=",",
        print_header=False,
        bigquery_conn_id=google_cloud_connection_id,
        dag=dag,
    )

奖励：在 BigQuery 中设置一个 tmp_dataset（或任何名称）作为临时数据集，并具有固定的表过期时间。

我只是建议如何解决您的确切问题，但请注意，这可能是了解表是否有（最近）数据的更好方法。例如，您还可以查询表元数据并获取有关最后一个分区的信息，然后根据需要管理此信息。