我想在DAG A 完成执行后清除DAG B中的任务。 A和B都是预定的 DAG。
I want to clear the tasks in DAG B when DAG A completes execution. Both A and B are scheduled DAGs.
是否有个操作符 /路清除任务状态并以编程方式重新运行DAG B?
Is there any operator/way to clear the state of tasks and re-run DAG B programmatically?
我知道 CLI选项和Web UI选项可清除任务。
I'm aware of the CLI option and Web UI option to clear the tasks.
推荐答案我建议在这里远离CLI!
I would recommend staying away from CLI here!
与通过BashOperator和/或CLI模块相比,引用对象时dags /任务的气流功能要好得多。
The airflow functionality of dags/tasks are much better exposed when referencing the objects, as compared to going through BashOperator and/or CLI module.
在上添加python操作dag A ,名为 clear_dag_b,它从 dags 文件夹(模块)中导入dag_b,并且该操作是:
Add a python operation to dag A named "clear_dag_b", that imports dag_b from the dags folder(module) and this:
from dags.dag_b import dag as dag_b def clear_dag_b(**context): exec_date = context[some date object, I forget the name] dag_b.clear(start_date=exec_date, end_date=exec_date)重要!如果您由于某种原因未将dag_b计划时间与开始日期/结束日期进行匹配或重叠,则clear()操作将丢失dag执行。此示例假设dag A 和 B 的安排相同,并且您只想从 B 清除日期 X ,当 A 执行第 X
Important! If you for some reason do not match or overlap the dag_b schedule time with start_date/end_date, the clear() operation will miss the dag executions. This example assumes dag A and B are scheduled identical, and that you only want to clear day X from B, when A executes day X
检查dag_b是否已经运行可能是有意义的是否在清除之前
It might make sense to include a check for whether the dag_b has already run or not, before clearing:
dab_b_run = dag_b.get_dagrun(exec_date) # returns None or a dag_run object更多推荐
以编程方式清除气流任务实例的状态
发布评论