具有不同调度程序间隔的 Airflow ExternalTask​​Sensor

Question

目前我有两个 DAG:DAG_A 和 DAG_B。两者都以 schedule_interval=timedelta(days=1)

运行

DAG_A 有一个 Task1，通常需要 7 个小时才能运行。而DAG_B只需要3个小时。

DAG_B 有一个 ExternalTask​​Sensor(external_dag_id="DAG_A", external_task_id="Task1") 但也使用每小时生成的一些其他信息 X。

增加 DAG_B 频率以使其每天至少运行 4 次的最佳方法是什么？据我所知，两个 DAG 必须具有相同的 schedule_interval。但是，我想尽可能多地更新 DAG_B 上的 X。

---

一种可能性是创建另一个具有 DAG_B 的 ExternalTask​​Sensor 的 DAG。但我认为这不是最好的方法。

Answer 1

如果我没理解错的话，你的条件是:

• 每天继续运行DAG_A
• 每天运行 DAG_B n 次
• 每次 DAG_B 运行时，它都会等待 DAG_A__Task_1 完成

我认为您可以通过指示 ExternalTask​​Sensor 等待 DAG_A 的预期执行日期来轻松调整您当前的设计。

来自ExternalTaskSensor运算符定义:

Waits for a different DAG or a task in a different DAG to complete for a specific execution_date

execution_date 可以使用 execution_date_fn 参数定义:

execution_date_fn (Optional[Callable]) – function that receives the current execution date as the first positional argument and optionally any number of keyword arguments available in the context dictionary, and returns the desired execution dates to query. Either execution_delta or execution_date_fn can be passed to ExternalTaskSensor, but not both.

你可以这样定义传感器:

    wait_for_dag_a = ExternalTaskSensor(
        task_id='wait_for_dag_a',
        external_task_id="external_task_1",
        external_dag_id='dag_a_id',
        allowed_states=['success', 'failed'],
        execution_date_fn=_get_execution_date_of_dag_a,
        poke_interval=30
    )

_get_execution_date_of_dag_a 使用 get_last_dagrun 对数据库执行查询，允许您获取 DAG_A 的最后一个 execution_date .

from airflow.utils.db import provide_session
from airflow.models.dag import get_last_dagrun

@provide_session
def _get_execution_date_of_dag_a(exec_date, session=None,  **kwargs):
    dag_a_last_run = get_last_dagrun(
        'dag_a_id', session)
    return dag_a_last_run.execution_date

我希望这种方法能帮到你。您可以在 this answer 中找到一个工作示例.