如何使用 ExternalTask​​Sensor 触发 Airflow DAG 独立运行

如何解决如何使用 ExternalTask​​Sensor 触发 Airflow DAG 独立运行

我构建了两个 DAG(dag_a,dag_b) 并在 dag_b 中创建了一个外部任务传感器，它会戳破 dag_a。这些 DAG 有两个用例：

1. 同时调度dag_a和dag_b，使用依赖先处理dag_a再处理dag_b
2. 独立手动触发 dag_b，无需关心 dag_a。

使用 ExternalTask​​Sensor，用例 1 工作得很好。但是用例 2 不起作用。并且 dag_b 将在 ExternalTask​​Sensor 处为 dag_a 停止并永远戳。有没有办法在某些条件下跳过 ExternalTask​​Sensor 并独立运行 dag_b？

解决方法

从 Airflow 2.0.0 开始，我认为没有必要使用 ExternalTaskSensor。使用 TriggerDagRunOperator，您可以创建和调度充当控制器的 DAG，有两个任务负责触发 DAG_A 和 DAG_B。

使用 wait_for_completion 参数，您可以实现第一个用例，而不会影响独立触发 DAG_B 的可能性。您还可以指定期望 poke_interval 和允许或失败状态。

Docs:

• wait_for_completion (bool) -- 是否等待 dag 运行完成。 （默认：False）
• poke_interval (int) -- 当 wait_for_completion=True 时戳间隔以检查 dag 运行状态。 （默认：60）
• allowed_states (list) -- 允许的状态列表，默认为 ['success']
• failed_states (list) -- 失败或不允许的状态列表，默认为无

这是指向 example DAG 随 Airflow 提供的链接。