如何在Airflow DAG中仅运行指定任务及其后续依赖

现有DAG配置 #

当前DAG固定依赖逻辑如下：

• 全量ML处理器任务列表定义：ml_processors = [a, b, c, d, e]
• 依赖流转规则：
  • abc_task >> ml_processors：前置abc_task执行成功后，a到e共5个ML模型任务并行启动
  • ml_processors >> xyz_task：所有5个ML处理器全部执行成功后，启动最终的xyz_task

核心需求 #

当单个或多个ML模型迭代到更高精度版本时，仅需要重跑对应已升级的模型任务、以及后续的xyz_task流程，不需要重复执行前置abc_task、以及未升级的其余ML处理器任务；如果同时升级c、d两个模型，希望一次性触发c_processor、d_processor >> xyz_task的链路，避免分两次单独执行流程。

已尝试方案的痛点 #

• 回溯历史DAG运行实例、清除指定任务记录重跑的方式：当需要同时重跑多个不同ML处理器的链路时，需要分步操作、重复触发多轮从处理器到xyz_task的流程，执行效率低
• 原生Airflow backfill能力：仅支持全DAG维度的历史数据回补，不支持选定部分任务的定向执行

当前环境情况 #

• 托管环境为Google Composer
• DAG触发方式为GCP存储桶文件上传事件触发

具体可行方案 #

方案1：批量重置历史运行实例的任务状态 #

不需要改动现有DAG代码，直接利用Airflow原生任务管理能力实现：

1. 定位到需要重跑的对应时间窗口的已成功DAG运行实例
2. 批量选中需要重跑的所有目标ML处理器任务（比如c、d），将其状态清除为未启动；不要改动abc_task和未升级的a、b、e任务的已成功状态
3. 触发选中的任务运行时，勾选「等待所有并行前置任务完成后再触发下游」选项
   • Airflow会自动识别上游已存在的成功状态，不会重跑abc_task；当c、d两个任务全部执行成功后，会自动触发依赖的xyz_task运行，不需要单独分步操作

方案2：带动态参数的DAG轻量改造 #

适合需要频繁定向重跑的场景，改造后无需每次操作历史任务状态：

1. 在DAG入口增加可传入参数enabled_processors，默认值为全量的ml_processors列表，完全兼容原有全量运行的触发逻辑
2. 改造依赖生成逻辑：仅将enabled_processors参数指定的处理器任务纳入执行依赖链；比如参数传入[c_processor, d_processor]时，实际生效的依赖会自动变为abc_task >> [c, d] >> xyz_task
3. 触发时，可通过GCS上传事件的元数据携带参数，或在DAG手动触发入口传入对应参数，即可一次性跳过不需要执行的任务，完成定向链路的全流程执行

方案3：分支节点隔离的低侵入改造 #

不对原有全量运行逻辑做大幅修改：

1. 保持原有全量依赖结构不变，在每个ML处理器任务与xyz_task之间增加轻量分支判断节点
2. 分支节点的判断逻辑为：仅当当前ML处理器属于本次重跑指定的升级任务列表时，才传递成功信号给xyz_task；否则直接跳过向下传递的流程
3. 将xyz_task的触发规则修改为“收到任意一个上游分支传递的成功信号即启动执行”
   • 日常全量运行时，所有分支节点都会传递信号，不影响原有执行逻辑；仅定向重跑时通过参数控制信号传递范围，即可实现定向链路执行

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