如何设置Airflow DAG调度让下游汇聚任务仅执行一次？

问题根因 #

你给出的Task1 >> [Task2, Task3] >> Task4依赖写法本身符合Airflow语法规范，默认触发规则下Task4仅会在Task2、Task3全部成功后执行1次，出现重复执行的现象通常是以下几个配置错误导致的：

修复方案 #

• 检查Task4的实例化逻辑，确保同一个task_id仅在DAG中实例化1次，不要把Task4的定义放在生成Task2、Task3的循环逻辑中，重复实例化相同task_id的任务会被Airflow识别为多个独立任务，分别关联上游后就会出现多次执行的情况。
• 显式指定Task4的trigger_rule为默认值all_success，避免继承全局配置或者其他任务的异常规则，配置示例如下：

from airflow.operators.bash import BashOperator

Task4 = BashOperator(
    task_id="task4",
    bash_command="echo '执行Task4逻辑'",
    trigger_rule="all_success", # 该规则要求所有上游任务全部成功后才触发当前任务，且仅触发1次
    dag=dag
)

• 如果你使用Airflow 2.x的动态任务映射功能，确认没有给Task4调用.expand()方法，多余的动态展开配置会生成多个Task4实例，每个实例分别响应上游的触发信号导致多次执行。
• 确认依赖关系没有被拆分为两行独立的链式声明，比如不要写成Task1 >> Task2 >> Task4 + Task1 >> Task3 >> Task4的形式，虽然逻辑上等价，但部分旧版本Airflow的调度器可能会识别错误，保持你当前的合并写法即可。
• 上述配置都确认无误的话，清空Airflow的DAG缓存，重启scheduler服务后再观察树视图即可。

内容的提问来源于stack exchange，提问作者박현균