要解决Airflow中DAG任务失败但被标记为成功的问题，可以使用以下方法：

1. 强制任务失败：
   • 在DAG中的失败任务之后添加一个fail任务，可以使用ShortCircuitOperator或PythonOperator来实现。
   • 例如，使用ShortCircuitOperator可以将任务设置为始终返回False，从而使其失败。

from airflow.operators.python_operator import ShortCircuitOperator

def check_failure():
    # 检查任务失败的条件
    # 返回False以表示任务失败
    return False

dag = DAG('my_dag', ...)
 
task1 = ...
task2 = ...
fail_task = ShortCircuitOperator(
    task_id='fail_task',
    python_callable=check_failure,
    dag=dag
)
 
task1 >> task2 >> fail_task

2. 使用trigger_rule属性：
   • 在DAG中的失败任务之后的依赖任务上设置trigger_rule='all_failed'属性。
   • 这将确保只有在所有前置任务都失败时，才会执行该任务。

from airflow.operators.dummy_operator import DummyOperator

dag = DAG('my_dag', ...)
 
task1 = ...
task2 = ...
task3 = DummyOperator(
    task_id='task3',
    trigger_rule='all_failed',  # 设置trigger_rule属性
    dag=dag
)
 
task1 >> task2 >> task3

这样，如果task2失败，task3将不会执行，并且DAG将被标记为失败。

请注意，这些方法都要求任务在实际运行时失败，而不是在DAG编译时失败。因此，如果任务在编译时失败，这些方法可能无法解决问题。在这种情况下，需要根据具体情况进行调试和修复。