S3ToRedshiftOperator 无法获取凭据

如何解决S3ToRedshiftOperator 无法获取凭据

我正在使用气流 2.x 并且我正在尝试使用 S3ToRedshiftOperator 将数据文件从 S3 移动到红移

    s3_to_redshift = S3ToRedshiftOperator(
        task_id = 'copy_from_S3_to_Redshift',aws_conn_id='S3_connection',redshift_conn_id='Redshift_conn',schema='public',table='department',s3_bucket='xxxxx',s3_key='department.csv'
    )

但它失败并抛出错误

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/airflow/models/taskinstance.py",line 1138,in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context,task)
  File "/usr/local/lib/python3.8/dist-packages/airflow/models/taskinstance.py",line 1311,in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context,task_copy)
  File "/usr/local/lib/python3.8/dist-packages/airflow/models/taskinstance.py",line 1341,in _execute_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python3.8/dist-packages/airflow/providers/amazon/aws/transfers/s3_to_redshift.py",line 111,in execute
    credentials = s3_hook.get_credentials()
  File "/usr/local/lib/python3.8/dist-packages/airflow/providers/amazon/aws/hooks/base_aws.py",line 478,in get_credentials
    return session.get_credentials().get_frozen_credentials()
AttributeError: 'nonetype' object has no attribute 'get_frozen_credentials'

我已经在 Airflow UI 的连接选项卡中创建了 S3 连接和 Redshift 连接。

注意：我的 redshift 在 EC2 集群上，我在 .ssh/config 文件中本地转发端口，然后我使用主机名 Localhost 进行连接。此过程适用于 Dbeaver 或任何数据库客户端。

任何帮助将不胜感激。