如何将外部 DAG 作为我的 DOG 的一部分运行？

Question

我是 Airflow 的新手，我正在尝试运行一个外部 DAG（由另一个团队开发和拥有），作为我的 DAG 流程的一部分。

我正在查看 SubDagOperator，但似乎出于某种原因它强制将 subdag 的名称设为。我不能这样做，因为 child dag 属于不同的团队。

这是我的代码示例：

parent_dag = DAG(
dag_id='parent_dag', default_args=args,
schedule_interval=None)

external_dag = SubDagOperator(
subdag=another_teams_dag,
task_id='external_dag',
dag=parent_dag,
trigger_rule=TriggerRule.ALL_DONE
)

而对方队的dag是这样定义的：

another_teams_dag = DAG(
dag_id='another_teams_dag', default_args=args,
schedule_interval=None)

但我收到此错误：

The subdag's dag_id should have the form '{parent_dag_id}.{this_task_id}'. Expected 'parent_dag.external_dag'; received 'another_teams_dag'.

有什么想法吗？我错过了什么？

Answer 1

使用TriggerDagRunOperator

更多信息：https://airflow.apache.org/code.html#airflow.operators.dagrun_operator.TriggerDagRunOperator

示例：

触发的 Dag：https://github.com/apache/incubator-airflow/blob/master/airflow/example_dags/example_trigger_controller_dag.py

被触发的 Dag: https://github.com/apache/incubator-airflow/blob/master/airflow/example_dags/example_trigger_target_dag.py

对于你的情况，你可以使用类似的东西：

trigger = TriggerDagRunOperator(task_id='external_dag',
                                trigger_dag_id="another_teams_dag",
                                dag=dag)

如何将外部 DAG 作为我的 DOG 的一部分运行？

How to runn external DAG as part of my DAG?

directed-acyclic-graphs

airflow