以编程方式清除气流任务实例的状态 [英] Programmatically clear the state of airflow task instances

问题描述

我想在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

这篇关于以编程方式清除气流任务实例的状态的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！