AWS MWAA配置aws_default连接后仍无法获取凭证如何解决

问题核心原因 #

你遇到的报错分为两个层面，首先要区分日志来源：

• 你看到的aws_default相关日志是MWAA平台本身拉取CloudWatch任务日志时输出的，不是你的DAG任务执行时产生的
• 你直接通过BashOperator执行AWS CLI命令时，Airflow不会自动将连接面板配置的aws_default凭证注入到Bash子进程中，AWS CLI只会按照自身默认的凭证链路查找凭证，自然读不到你在Airflow连接里配置的内容

排查步骤（先确认配置正确性） #

1. 打开Airflow连接面板，检查aws_default连接的配置：
   • 连接类型必须选择 Amazon Web Services，选错类型会导致Airflow无法正确读取连接字段
   • 若你要手动配置静态凭证：登录名填AWS_ACCESS_KEY_ID，密码填AWS_SECRET_ACCESS_KEY，额外参数中填入{"region_name": "你的AWS区域"}
   • 若你要使用MWAA默认的执行角色权限：把aws_default连接的所有字段清空即可，MWAA会自动将执行角色的权限注入到aws_default连接中，手动填写错误配置反而会覆盖默认逻辑导致无法获取凭证
2. 确认MWAA的执行角色已经附加了对应操作的权限，比如你要启动DMS任务，需要给执行角色加dms:StartReplicationTask的权限

两种解决方案 #

方案1（最推荐）：使用Airflow官方AWS Operator #

不要用BashOperator直接跑CLI，改用AWS Provider提供的官方Operator，会自动读取aws_default连接的凭证，无需手动处理凭证逻辑：

1. 先在MWAA的requirements.txt中添加依赖（对应你的Airflow版本选择合适的包版本）：

apache-airflow-providers-amazon>=7.0.0

2. 修改DAG代码如下：

from datetime import datetime, timedelta
import pendulum
from airflow import DAG
from airflow.providers.amazon.aws.operators.dms import DmsStartReplicationTaskOperator
from airflow.utils.dates import days_ago

local_tz = pendulum.timezone("America/New_York")
start_date = datetime(2021, 11, 9, 8, tzinfo=local_tz)

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}
with DAG(
    'dms-postgres-dialog-label-pg-test',
    default_args=default_args,
    description='',
    schedule_interval=timedelta(days=1),
    start_date=start_date,
    tags=['example'],
) as dag:

    t1 = DmsStartReplicationTaskOperator(
        task_id='start-replication-task',
        replication_task_arn='arn:aws:dms:us-east-1:blah',
        start_replication_task_type='reload-target',
        aws_conn_id='aws_default', # 可以省略，默认就是aws_default
    )

方案2：坚持使用BashOperator的处理方式 #

手动将aws_default连接的凭证注入到Bash进程的环境变量中：

from datetime import datetime, timedelta
import pendulum
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.hooks.base import BaseHook

local_tz = pendulum.timezone("America/New_York")
start_date = datetime(2021, 11, 9, 8, tzinfo=local_tz)

aws_conn = BaseHook.get_connection('aws_default')
aws_env = {
    "AWS_ACCESS_KEY_ID": aws_conn.login,
    "AWS_SECRET_ACCESS_KEY": aws_conn.password,
    "AWS_DEFAULT_REGION": aws_conn.extra_dejson.get('region_name', 'us-east-1')
}

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}
with DAG(
    'dms-postgres-dialog-label-pg-test',
    default_args=default_args,
    description='',
    schedule_interval=timedelta(days=1),
    start_date=start_date,
    tags=['example'],
) as dag:

    t1 = BashOperator(
        task_id='start-replication-task',
        bash_command="""
        aws dms start-replication-task --replication-task-arn arn:aws:dms:us-east-1:blah --start-replication-task-type reload-target
        """,
        env=aws_env
    )

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