如何将 DAG 运行日期传递给任务？

Question

我是气流新手，想弄清楚如何将 DAG 运行日期传递给每个任务，我的 DAG 中有这个：

tzinfo=tz.gettz('America/Los_Angeles')
dag_run_date = datetime.now(_tzinfo)

dag = DAG(
    'myDag', 
    default_args=default_args,
    schedule_interval = None,
    params = {
        "runDateTimeTz" : dag_run_date.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
    }
)

然后我尝试将 runDateTimeTz 参数传递给我的每个任务，就像这样..

task1 = GKEPodOperator(
    image='gcr.io/myJar:1.0.1.45',           
    cmds=['java'],
    arguments=["-jar","myJar.jar", {{params.runDateTimeTz}}"],
    dag=dag)

task2 = GKEPodOperator(
    image='gcr.io/myJar2:1.0.1.45',           
    cmds=['java'],
    arguments=["-jar","myJar2.jar", {{params.runDateTimeTz}}"],
    dag=dag)

我的任务已正确执行，但我希望它们都在 params.runDateTimeTz 中收到相同的运行日期，但它没有发生，例如 task1 得到 params.runDateTimeTz=2020-04-16T07:42:47.412716-07:00 而 task2 得到 params.runDateTimeTz= 2020-04-16T07:43:29.913289-07:00

我想这种行为与气流填充 DAG 的 params 的方式有关，看起来 params.runDateTimeTz 仅在任务开始运行时才获得，但我想在之前获得它并将其发送到每个任务作为一个参数，期望所有任务都获得相同的值。

有人可以帮助我解决我做错了什么吗？

Answer 1

您可以使用 Airflow Macros 中的 execution_date 或 ds：

详情：https://airflow.apache.org/docs/stable/macros-ref#default-variables

task1 = GKEPodOperator(
    image='gcr.io/myJar:1.0.1.45',           
    cmds=['java'],
    arguments=["-jar","myJar.jar", "{{ ds }}"],
    dag=dag)

task2 = GKEPodOperator(
    image='gcr.io/myJar2:1.0.1.45',           
    cmds=['java'],
    arguments=["-jar","myJar2.jar", "{{ ds }}"],
    dag=dag)

如果您需要时间戳，可以使用{{ ts }}