将新任务引入DAG时,气流调度程序是否会重新运行先前的DAG运行?
问题描述:
我有一个DAG定义,动态地将任务从配置到DAG,例如:将新任务引入DAG时,气流调度程序是否会重新运行先前的DAG运行?
for k, v in config.iteritems():
bash_task = BashOperator(task_id='{}_task'.format(k), ...)
...
如果我增加更多的项目config
,将气流调度重复运行所有的以前的工作,否则我将不得不手动(通过运行airflow scheduler
或airflow backfill
)?
答
调度程序不会检查所有以前的DagRun的所有TaskInstances。如果在相关日期范围内运行airflow backfill
,则应该选取所有未完成的任务。
警告:没有设置气流,从DAG中删除任务。您的数据库可能会保留对不再有效任务ID的引用,并且Airflow可能会拒绝对该DAG执行任何操作,直到任务恢复或DAG中的所有其他TaskInstances被删除。