如何在Airflow DAG内主动触发任务失败且不出现Broken DAG报错

Broken DAG报错原因 #

你遇到的Broken DAG报错和主动抛出异常的逻辑无关，是代码本身存在运行时就会触发的错误，Airflow Scheduler在解析DAG文件加载自定义逻辑时提前触发了报错，核心问题有3个：

• 执行SQL后直接调用cursor.fetchmany(10)：如果执行的是INSERT/UPDATE/DELETE/DDL等无结果集的SQL，这个操作会直接抛出「无结果可获取」的数据库异常，直接中断代码加载
• 最后一行日志语句存在格式化错误："Select process is completed".format(self.dest_table) 字符串中没有{}占位符，调用format方法会触发参数不匹配报错
• 未做依赖导入校验：如果没有提前导入PostgresHook、日志模块，或者自定义Operator没有正确继承BaseOperator，也会触发Broken DAG

正确实现代码 #

要主动触发DAG任务失败，建议抛出Airflow原生的AirflowException，Airflow会直接将任务标记为失败状态，不会产生额外解析问题，修改后代码如下：

# 提前导入需要的依赖
import logging
from airflow.models import BaseOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.exceptions import AirflowException

log = logging.getLogger(__name__)

# 你的自定义Operator类，必须继承BaseOperator
class YourCustomOperator(BaseOperator):
    # 这里保留你原来的__init__方法
    
    def execute(self, context):
        self.hook = PostgresHook(postgres_conn_id=self.dest_redshift_conn_id)
        conn = self.hook.get_conn()
        cursor = conn.cursor()
        log.info("Connected with " + self.dest_redshift_conn_id)
        log.info('Starting Query')
        print('Starting Query \n')
        cursor.execute("begin transaction;")
        sql_comm = self.read_queries_file()
        for command in sql_comm:
            sql_statement = command.strip()
            if sql_statement:
                log.info(sql_statement)
                log.info('\n')
                cursor.execute(sql_statement)
                # 只对查询类SQL尝试获取结果，避免无结果集报错
                if sql_statement.lower().startswith('select') or sql_statement.lower().startswith('with'):
                    res = cursor.fetchmany(10)
                    if cursor.rowcount > 0:
                        # 抛Airflow原生异常标记任务失败
                        raise AirflowException(f"Query returned results: {res}")
                    else:
                        log.info('\nNo result found in the query')
                else:
                    log.info(f"Non-query SQL executed, affected rows: {cursor.rowcount}")
        cursor.execute(" end transaction;")
        cursor.close()
        # 修复日志格式化错误
        log.info(f"Select process is completed for {self.dest_table}")

额外校验点 #

部署前可以先在本地直接运行DAG文件做语法校验，执行python your_dag_file.py，没有报错再部署到Airflow节点，就能避免Broken DAG问题。

内容的提问来源于stack exchange，提问作者Alan Mil