AWS MWAA 2.9.2中DBT命令无法执行问题求助

问题描述 #

因项目安全要求，我们将AWS MWAA中的Airflow版本从2.0.2升级至2.9.2，但升级后所有DBT命令均无法正常执行。测试DAG中的Python任务可正常运行，但通过BashOperator调用的DBT命令执行失败；尝试使用DbtRunOperator、DbtTestOperator也出现同样问题，且错误日志无有效排查信息。本地环境运行DBT命令完全正常，仅云端MWAA环境无法执行。

测试DAG代码 #

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python_operator import PythonOperator
from airflow.utils.task_group import TaskGroup
from airflow.utils.dates import days_ago

from airflow_dbt.operators.dbt_operator import (
    DbtRunOperator,
    DbtTestOperator
)
import os 
from datetime import timedelta, datetime
import pendulum
from datetime import datetime
import uuid

local_tz = pendulum.timezone("Europe/Paris")

DBT_FOLDER = 'Project/transformation/dbt'
DBT_BIN = '/usr/local/airflow/.local/bin/dbt'
DBT_PROJECT_DIR = f'/usr/local/airflow/dags/{DBT_FOLDER}/'
DBT_PROFILES_DIR = f'/usr/local/airflow/dags/{DBT_FOLDER}/'

ENV = os.environ.get("AIRFLOW__ENVIRONMENT")

ACCOUNT = ****
DATABASE = ****
ROLE = ****
WAREHOUSE = ****
USER = os.getenv('AIRFLOW__SECRET_VAR_SNOWFLAKE_USER')
PASSWORD = os.getenv('AIRFLOW__SECRET_VAR_SNOWFLAKE_PASSWORD')

if ENV.lower() == 'dev':
    DBT_TARGET = 'dev_live'
elif ENV.lower() == 'uat':
    DBT_TARGET = 'uat_live'
elif ENV.lower() == 'prod':
    DBT_TARGET = 'prod_live'
else:
    DBT_TARGET = ''

def print_hello():
    print("Hello from Airflow!")

def print_end():
    print("Hello from Airflow!- Ending")

default_args = {
    'owner': 'Project',
    'start_date': datetime(2022, 3, 9, 0, 15, tzinfo=local_tz),
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'Project_TRY',
    default_args=default_args,
    description='Project Try',
    # schedule_interval='10 */2 * * *',
    schedule_interval=None,
    catchup=False,
    max_active_runs=1,
)

with dag:
    start_task = PythonOperator(
        task_id='print_hello_task',
        python_callable=print_hello,
    )

    try_dbtcommand = BashOperator(
        task_id='try_dbtcommand',
        bash_command=f"{DBT_BIN} run --project-dir {DBT_PROJECT_DIR} --profiles-dir {DBT_PROFILES_DIR} --target {DBT_TARGET} --select model_try"
    )
    end_task=PythonOperator(
        task_id='print_end_task',
        python_callable=print_end,
    )

start_task >> try_dbtcommand >> end_task

依赖包配置 #

apache-airflow==2.9.2
airflow-dbt==0.4.0
boto3==1.34.106
dbt-snowflake==1.5.4
sqlfluff==2.3.5
sqlfluff-templater-dbt==2.3.5

使用Airflow 2.9.2对应的Python 3.11版本约束文件。

排查建议 #

• 路径与权限验证
  • 确认MWAA环境中DBT_BIN路径是否正确：在MWAA任务日志中执行which dbt验证实际可执行文件位置，避免硬编码路径导致的环境差异
  • 检查DBT项目目录（DBT_PROJECT_DIR）的权限，MWAA运行用户需具备读取该目录下dbt_project.yml、profiles.yml及模型文件的权限
• 依赖兼容性
  • airflow-dbt==0.4.0版本较旧，与Airflow 2.9.2可能存在兼容性问题，建议替换为维护活跃的airflow-dbt-python库（原airflow-dbt已停止维护）
  • 在本地使用相同约束文件执行pip check，排查dbt-snowflake==1.5.4与Airflow 2.9.2的依赖冲突
• 环境变量与配置
  • 确认MWAA环境中AIRFLOW__ENVIRONMENT、AIRFLOW__SECRET_VAR_SNOWFLAKE_USER、AIRFLOW__SECRET_VAR_SNOWFLAKE_PASSWORD等变量已正确配置并可被任务读取
  • 检查profiles.yml中的目标配置（dev_live/uat_live/prod_live）是否与MWAA环境的Snowflake权限匹配，可在任务中添加dbt debug命令输出详细连接日志
• 任务执行环境
  • 修改BashOperator命令，先执行debug再运行任务，获取详细日志：

    {DBT_BIN} debug --project-dir {DBT_PROJECT_DIR} --profiles-dir {DBT_PROFILES_DIR} --target {DBT_TARGET} && {DBT_BIN} run --project-dir {DBT_PROJECT_DIR} --profiles-dir {DBT_PROFILES_DIR} --target {DBT_TARGET} --select model_try
    

  • 确认MWAA执行角色具备访问Snowflake的权限，以及存放DAG与DBT项目的S3存储桶的读写权限

内容的提问来源于stack exchange，提问作者Sri Harsha Damarla