DAG导入错误：PostgresToGCSOperator传入无效参数求助

Docker部署Airflow时，尝试将本地PostgreSQL数据导入Google Cloud Storage，遇到DAG导入错误，报错提示PostgresToGCSOperator（task_id: postgres_to_gcs）传入了无效参数google_cloud_storage_conn_id。

报错栈信息 #

Broken DAG: [/opt/airflow/dags/postgres_to_bigquery.py] Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 408, in apply_defaults
    result = func(self, **kwargs, default_args=default_args)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 756, in __init__
    f"Invalid arguments were passed to {self.__class__.__name__} (task_id: {task_id}). "
airflow.exceptions.AirflowException: Invalid arguments were passed to PostgresToGCSOperator (task_id: postgres_to_gcs). Invalid arguments were:
**kwargs: {'google_cloud_storage_conn_id': 'gcp_conn'}

相关代码片段 #

GCS_CONN = Variable.get('GCS_CONN')

default_args={
    'owner': 'airflow',
    'retries': 0,
    'retry_delay': timedelta(minutes=5),
    }

with DAG(
    dag_id = 'postgres_to_bigquery',
    default_args = default_args,
    start_date = datetime(2022, 10, 3),
    schedule_interval = '@once'
) as dag:

    start = DummyOperator(
        task_id = 'start',

    )

    postgres_to_gcs = PostgresToGCSOperator(
        task_id = f'postgres_to_gcs',
        postgres_conn_id = 'postgres_localhost',
        sql = f'select * from orders;',
        bucket = 'airflow_fakri',
        filename = f'airflow_fakri/data/orders.csv',
        export_format = 'csv',
        gzip = False,
        use_server_side_cursor = False,
        google_cloud_storage_conn_id = GCS_CONN

    )

解决方法 #

1. 修正参数名称：PostgresToGCSOperator中用于指定GCP连接的参数是gcp_conn_id，而非google_cloud_storage_conn_id，修改代码中的参数名：

   postgres_to_gcs = PostgresToGCSOperator(
       # 其他参数保持不变
       gcp_conn_id = GCS_CONN
   )
   

2. 验证GCP连接配置：确保Airflow UI中已创建ID为GCS_CONN的GCP连接，且该连接包含有效的服务账号密钥、项目ID等信息。
3. 重新加载DAG：重启Airflow调度器和Webserver服务，让修改后的DAG重新加载。

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