Airflow的DbtCloudRunJobOperator无法识别DBT Cloud任务成功完成，如何解决？

问题描述 #

在Google Cloud Composer（Airflow 2.7.3，Composer 2.7.1）环境中，使用DbtCloudRunJobOperator（版本3.8.0）触发dbt Cloud任务时，dbt Cloud任务本身执行成功且无异常，但Airflow算子始终将任务标记为失败，该问题与任务配置或dbt Cloud项目设置无关。

使用的算子代码如下：

trigger_dbt_job = DbtCloudRunJobOperator(
    task_id="trigger_dbt_job",
    job_id=job_id,
    wait_for_termination=True,  # Flag to wait on a job run’s termination.
    check_interval=30, # The interval in seconds between each check for termination.
    execution_timeout=timedelta(minutes=60), # The maximum amount of time to wait for the job to complete.
    steps_override=steps_override,
    retries=config.get(
        "max_retries", 1
    ),  # If max_retries is found in the config, it uses that value. Otherwise it uses 1.
)

预期结果：Airflow算子应正确识别dbt Cloud任务的成功状态，将自身标记为success，确保DAG后续任务正常执行。

排查方向与解决方法 #

1. 验证dbt Cloud API权限与响应状态 #

• 确认Airflow使用的dbt Cloud API Token具备read和run权限，避免因权限不足导致状态获取异常。
• 手动调用dbt Cloud的Get Run API（GET /api/v2/accounts/{account_id}/runs/{run_id}/），检查返回的status字段值：dbt Cloud任务成功的标准状态码为10，若算子对状态的判断逻辑与该值不匹配，会导致误判。
• 查看Airflow任务日志中关于状态检查的输出，确认算子获取的状态值是否符合预期。

2. 修复版本兼容性问题 #

当前使用的apache-airflow-providers-dbt-cloud 3.8.0版本与Airflow 2.7.3可能存在兼容性问题：

• 尝试将provider版本降级至3.7.0，或升级至最新兼容版本，验证状态识别是否恢复正常。
• 查阅该provider的发布记录，确认是否存在针对任务状态判断逻辑的修复内容。

3. 自定义状态检查逻辑（替代方案） #

若算子原生的状态判断存在缺陷，可通过扩展算子或直接使用DbtCloudHook手动实现状态检查：

from airflow.providers.dbt.cloud.hooks.dbt import DbtCloudHook
from airflow.models.baseoperator import BaseOperator
from airflow.utils.decorators import apply_defaults
import time

class CustomDbtCloudRunJobOperator(BaseOperator):
    @apply_defaults
    def __init__(self, job_id, dbt_cloud_conn_id="dbt_cloud_default", check_interval=30, execution_timeout=None, **kwargs):
        super().__init__(**kwargs)
        self.job_id = job_id
        self.dbt_cloud_conn_id = dbt_cloud_conn_id
        self.check_interval = check_interval
        self.execution_timeout = execution_timeout

    def execute(self, context):
        hook = DbtCloudHook(dbt_cloud_conn_id=self.dbt_cloud_conn_id)
        run_info = hook.trigger_job_run(job_id=self.job_id)
        run_id = run_info["data"]["id"]
        
        # 自定义状态检查逻辑，覆盖算子原生判断
        status = hook.get_job_run_status(run_id)
        start_time = time.time()
        
        while status not in [10, 20, 30]:  # 10=成功, 20=失败, 30=取消
            if self.execution_timeout and (time.time() - start_time) > self.execution_timeout.total_seconds():
                raise TimeoutError("等待dbt任务完成超时")
                
            self.log.info(f"当前dbt任务状态: {status}, 等待中...")
            time.sleep(self.check_interval)
            status = hook.get_job_run_status(run_id)
        
        if status != 10:
            raise Exception(f"dbt Cloud任务 {run_id} 执行失败，状态码: {status}")
        self.log.info(f"dbt Cloud任务 {run_id} 执行成功")

4. 调整任务超时设置 #

检查execution_timeout参数是否合理：若dbt Cloud任务实际执行时间接近或超过该值，Airflow会提前终止任务并标记为失败。可延长超时时间至合理范围，或暂时取消该设置进行测试。

5. 深度排查日志 #

查看Airflow任务的完整日志，重点关注以下内容：

• 算子获取dbt Cloud任务状态的日志行，确认状态值是否正确传递。
• 是否存在API调用失败、JSON解析错误等隐藏异常，这些问题可能导致算子误判任务状态。

内容的提问来源于stack exchange，提问作者Davy van der Horst