如何在Airflow中获得DAG链执行时间？

Question

假设我有两个DAG，其中dag2使用TriggerDagRunOperator执行dag1作为其流程的一部分，如下所示：

• Dag 1：Tusk 1>小睡一下>闲逛
• dag 2：任务4> dag1>任务5

现在让我们说dag2每天下午5点安排一次。有没有办法在我运行dag1时获取dag2（父DAG）的执行时间戳？是否有任何内置参数保存该值？

如果发生了某些事情并且dag2比平时更晚被触发，那么让我们说当天下午6点，那么我仍然希望得到原始的调度时间 - 也就是我在dag1时的下午5点。

Answer 1

将函数传递给python_callable的TriggerDagRunOperator参数，该参数将execution_date注入触发的DAG：

def inject_execution_date(context, dag_run_obj):
  dag_run_obj.payload = {"parent_execution_date": context["execution_date"]}
  return dag_run_obj

[...]

trigger_dro = TriggerDagRunOperator(python_callable=inject_execution_date, [...])

您可以使用context["conf"]["parent_execution_date"]在子DAG中访问它