如何获取所有任务的列表及其在气流中运行的当前DAG的状态

Question

需要帮助，以提取所有任务的列表以及当前DAG运行的当前状态[成功/失败]。 我有一个Python操作员的任务，该操作员在工作流程结束时执行。此任务的责任是返回以状态执行的任务。

Answer 1

您可以创建一个pythOnoperator，该操作员从DAG_RUN读取所有任务。

task_id =“ tasks”的XCOM是：

with DAG(
    dag_id="get_tasks",
    description="get tasks",
    schedule_interval=None,
    start_date=datetime.datetime(2021, 1, 1),
    tags=["tasks"],
) as dag:
    start_dag = EmptyOperator(task_id="start")
    end_dag = EmptyOperator(task_id="end")


    def get_tasks(**context):
        dagrun: DAG = context["dag_run"]
        tasks = {}
        for ti in dagrun.get_task_instances():
            tasks[ti.task_id] = ti.state
        return tasks

    tasks = PythonOperator(
        task_id="tasks",
        python_callable=get_tasks,
    )

    start_dag >> tasks >> end_dag

Answer 2

您可以使用气流API提取有关工作流的信息。您可以读取REST API文档在这里。例如，要列出特定dag的所有DAG运行：

http://<AIRFLOW_IP>:8080/api/v1/dags/{dag_id}/dagRuns

之后，您可以列出此特定DAG运行的所有任务：

https://<AIRFLOW_IP>:8080/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances

in airflow UI＆gt;＆gt;文档＆gt;＆gt; REST API参考（Swagger UI）您可以使用Swagger文档访问和测试API。