GCP Cloud Composer中使用BigQueryToPostgresOperator遇导入错误求助

在GCP Composer工作流（DAG）中尝试使用BigQueryToPostgresOperator时，触发以下导入错误：

ImportError: cannot import name 'bigquery_to_postgres' from 'airflow.providers.google.cloud.transfers' (/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/transfers/init.py)

当前使用的导入语句：

from airflow.providers.google.cloud.transfers import bigquery_to_postgres

Composer环境版本：

composer-1.20.12-airflow-2.4.3

原因 #

你使用的Composer环境对应Airflow 2.4.3，配套的apache-airflow-providers-google版本为8.10.0，而BigQueryToPostgresOperator是在Google Provider版本9.0.0及以上才新增的模块，因此当前环境中不存在该组件，导致导入失败。

解决办法 #

1. 升级Composer环境 #

将Composer环境升级到包含apache-airflow-providers-google>=9.0.0的版本，推荐选择：

• Composer 2版本（稳定性更高，支持更新的Provider组件）
• 或Composer 1的更高版本，如composer-1.21.1-airflow-2.4.3（需提前确认对应Provider版本）

升级完成后，使用正确的导入语句：

from airflow.providers.google.cloud.transfers.bigquery_to_postgres import BigQueryToPostgresOperator

2. 不升级环境的替代方案 #

如果无法升级环境，可通过两步转移实现数据同步：

第一步：BigQuery数据导出到GCS

使用BigQueryToCloudStorageOperator完成导出：

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

export_bq_to_gcs = BigQueryToCloudStorageOperator(
    task_id="export_bq_to_gcs",
    source_project_dataset_table="你的项目ID.数据集ID.表名",
    destination_cloud_storage_uris=["gs://你的存储桶/文件路径/导出文件.csv"],
    export_format="CSV",
    field_delimiter=",",
    print_header=True,
)

第二步：GCS数据导入Postgres

使用GCSToPostgresOperator完成导入：

from airflow.providers.google.cloud.transfers.gcs_to_postgres import GCSToPostgresOperator

import_gcs_to_postgres = GCSToPostgresOperator(
    task_id="import_gcs_to_postgres",
    bucket="你的存储桶",
    object_name="文件路径/导出文件.csv",
    postgres_conn_id="你的Postgres连接ID",
    table="Postgres目标表名",
    schema="Postgres目标模式（可选）",
    delimiter=",",
    skip_leading_rows=1,
)

# 设置任务依赖
export_bq_to_gcs >> import_gcs_to_postgres

也可以通过PythonOperator编写自定义逻辑，直接用BigQuery客户端读取数据后写入Postgres，灵活性更高。

内容的提问来源于stack exchange，提问作者p.magalhaes