AWS MWAA集成Astronomer Cosmos：如何从Secrets Manager传递Snowflake凭证

核心逻辑 #

dbt的profiles.yml读取环境变量时，必须确保变量注入到dbt任务的运行环境中，而非仅在DAG定义层或Airflow全局设置。以下是具体实现步骤：

1. 验证Secrets Manager访问权限 #

确认MWAA执行角色已配置secretsmanager:GetSecretValue权限，否则无法成功拉取凭证。

2. 封装凭证获取函数 #

把凭证读取逻辑改成可复用的函数：

import boto3
import json

def fetch_snowflake_secrets(secret_arn):
    sm_client = boto3.client("secretsmanager")
    secret_response = sm_client.get_secret_value(SecretId=secret_arn)
    return json.loads(secret_response["SecretString"])

3. 在Cosmos任务中注入环境变量 #

有两种方式传递凭证，按需选择：

方式1：直接通过ProfileMapping传递参数（无需依赖profiles.yml的环境变量）

from datetime import datetime
from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping

# 拉取凭证
snowflake_creds = fetch_snowflake_secrets("arn:aws:secretsmanager:us-east-1:1234567890:secret:snowflake-dbt-creds")

profile_config = ProfileConfig(
    profile_name="snowflake",
    target_name="dev",
    profile_mapping=SnowflakeUserPasswordProfileMapping(
        profile_args={
            "account": snowflake_creds["account"],
            "user": snowflake_creds["user"],
            "password": snowflake_creds["password"],
            "warehouse": snowflake_creds["warehouse"],
            "database": snowflake_creds["database"],
            "schema": snowflake_creds["schema"],
            "role": snowflake_creds["role"],
        }
    )
)

with DAG(
    dag_id="dbt_snowflake_etl",
    schedule_interval="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    dbt_task_group = DbtTaskGroup(
        project_config=ProjectConfig("/usr/local/airflow/dags/dbt_project"),
        profile_config=profile_config,
        operator_args={"install_deps": True},
    )

方式2：通过ExecutionConfig设置环境变量（适配profiles.yml的env_var引用）

from datetime import datetime
from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig, ExecutionConfig

# 拉取凭证
snowflake_creds = fetch_snowflake_secrets("arn:aws:secretsmanager:us-east-1:1234567890:secret:snowflake-dbt-creds")

execution_config = ExecutionConfig(
    env={
        "DBT_SNOWFLAKE_ACCOUNT": snowflake_creds["account"],
        "DBT_SNOWFLAKE_USER": snowflake_creds["user"],
        "DBT_SNOWFLAKE_PASSWORD": snowflake_creds["password"],
        "DBT_SNOWFLAKE_WAREHOUSE": snowflake_creds["warehouse"],
        "DBT_SNOWFLAKE_DATABASE": snowflake_creds["database"],
        "DBT_SNOWFLAKE_SCHEMA": snowflake_creds["schema"],
        "DBT_SNOWFLAKE_ROLE": snowflake_creds["role"],
    }
)

with DAG(
    dag_id="dbt_snowflake_etl",
    schedule_interval="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    dbt_task_group = DbtTaskGroup(
        project_config=ProjectConfig("/usr/local/airflow/dags/dbt_project"),
        profile_config=ProfileConfig(profile_name="snowflake", target_name="dev"),
        execution_config=execution_config,
        operator_args={"install_deps": True},
    )

4. 匹配profiles.yml的环境变量引用 #

若使用方式2，确保profiles.yml内容如下：

snowflake:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: "{{ env_var('DBT_SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('DBT_SNOWFLAKE_USER') }}"
      password: "{{ env_var('DBT_SNOWFLAKE_PASSWORD') }}"
      warehouse: "{{ env_var('DBT_SNOWFLAKE_WAREHOUSE') }}"
      database: "{{ env_var('DBT_SNOWFLAKE_DATABASE') }}"
      schema: "{{ env_var('DBT_SNOWFLAKE_SCHEMA') }}"
      role: "{{ env_var('DBT_SNOWFLAKE_ROLE') }}"

5. 避坑提示 #

• 不要在Airflow全局环境变量中存储敏感凭证，存在泄露风险且无法按任务隔离
• 确认MWAA worker所在VPC已配置Secrets Manager的端点，安全组放行相关流量
• 若使用自定义DbtLocalOperator，需手动传递env参数，Cosmos的DbtTaskGroup会自动处理

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