抓取 dags 文件夹以提取 ExternalTaskSensor 任务和参数答案

【问题标题】：Scrape dags folder to extract ExternalTaskSensor tasks and parameters抓取 dags 文件夹以提取 ExternalTaskSensor 任务和参数
【发布时间】：2020-07-22 06:47:57
【问题描述】：

我们已经广泛使用 [ExternalTaskSensor][1]，以至于难以跟踪跨 dag 依赖项的数量。因此，我们想要一种方法来提取使用此传感器的所有任务以及传递给这些任务的参数，例如external_dag_id 和external_task_id。提取此信息将允许我们创建一个依赖项列表（如果需要，还可以创建一个图表）。

方法：到目前为止，我们已经能够使用list_dags cli 选项来获取所有 dag 的列表。然后，对于每个 dag，我们使用 -t 参数运行 list_tasks 选项以获取任务列表和使用的运算符。下一步是检索传递给这些任务的参数，这就是我们卡住的地方。是否有任何官方或非官方的方法来抓取这些数据？

信息：我们正在运行 Airflow 1.10.9 和 Composer 1.11.0。到目前为止，我们的脚本是用 python3 编写的。 [1]：https://airflow.readthedocs.io/en/stable/_modules/airflow/sensors/external_task_sensor.html

【问题讨论】：

标签： airflow google-cloud-composer

【解决方案1】：

你可以这样做：

dag_models = session.query(DagModel).filter(DagModel.is_active.is_(True)).all()

for dag_model in dag_models:
     dag = dag_model.get_dag()
     for task in dag.task_dict.values():
         if isinstance(task, ExternalTaskSensor):
             do_smth(task.external_dag_id, task.external_task_id)

【讨论】：

【解决方案2】：

您可以为此利用Airflow's metadb。

直接查询

SELECT operator
FROM task_instance
WHERE dag_id = 'my_dag'
  AND task_id = 'my_task';```

或使用SQLAlchemy

from airflow.utils.session import provide_session
from airflow.models import TaskInstance

@provide_session
def get_operator_name(my_dag_id: str, my_task_id: str, session=None) -> str:
    """Fetch TaskInstance from the database using pickling"""
    task_instance: TaskInstance = session.query(TaskInstance).filter(TaskInstance.dag_id == my_dag_id).filter(TaskInstance.task_id == my_task_id).first()
    return task_instance.operator

这种方法的缺点是它在task 至少运行一次后才能工作（并且它的条目已在TaskInstance 表中创建）

参考

cli.py

【讨论】：

task_instance 元数据表不包含有关传递给 ExternalTaskSensor 的 external_dag_id 和 external_task_id 的数据。