如何避免Airflow中DAG参数动态执行，使时间戳全局值一致？

问题根源 #

你当前的写法里，datetime.now()是在DAG文件被Airflow调度器解析时执行的，而调度器会定期重新解析DAG（默认30秒一次），导致不同任务甚至同一任务的不同运行拿到的时间戳不一致。如果要让同一个DAG运行实例（DAG Run）里的所有任务共享同一个时间戳，有以下几种可行方案：

方案1：预先计算时间戳并复用（DAG解析时固定） #

如果需要的是DAG被加载时的固定时间戳，直接在DAG定义前预先计算好这个值，再传入default_args，这样整个DAG生命周期内（直到下一次解析）这个值不会变：

from datetime import datetime
from airflow.decorators import dag, task
from airflow.operators.python import get_current_context

# 只在DAG解析时执行一次，固定时间戳
FIXED_TIMESTAMP = datetime.now()

default_dag_args = {
    'arg1': 'arg1-value',
    'arg2': 'arg2-value',
    'now': FIXED_TIMESTAMP
}

@dag(default_args=default_dag_args, schedule_interval='@daily', start_date=datetime(2023, 1, 1))
def my_dag():
    @task
    def python_task():
        context = get_current_context()
        now = context['dag'].default_args['now']
        print(now)
    
    python_task()

my_dag()

方案2：使用Airflow内置执行时间变量（推荐，DAG Run级固定） #

如果只需要和DAG运行绑定的基准时间（比如调度时间、数据区间起始时间），直接用Airflow内置的模板变量，TaskFlow API会自动注入这些参数，同一个DAG Run里所有任务拿到的值完全一致：

from datetime import datetime
from airflow.decorators import dag, task

@dag(schedule_interval='@daily', start_date=datetime(2023, 1, 1))
def my_dag():
    @task
    def python_task(execution_date, data_interval_start):
        # execution_date：DAG的调度执行时间
        # data_interval_start：当前数据周期的起始时间
        print(f"执行时间：{execution_date}")
        print(f"数据周期起始：{data_interval_start}")
    
    # 无需手动传参，TaskFlow自动注入
    python_task()

my_dag()

方案3：前置任务生成时间戳，跨任务传递（自定义DAG Run级时间） #

如果需要的是DAG Run启动时的实时时间（比如手动触发时的当前时间），可以用一个前置任务生成时间戳，然后传递给所有后续任务，确保同一个DAG Run里所有任务复用这个值：

from datetime import datetime
from airflow.decorators import dag, task

@dag(schedule_interval='@daily', start_date=datetime(2023, 1, 1))
def my_dag():
    @task
    def generate_run_timestamp():
        # 只在DAG Run启动时执行一次
        return datetime.now()
    
    @task
    def python_task(now):
        print(f"全局时间戳：{now}")
    
    # 生成时间戳并传递给所有需要的任务
    run_timestamp = generate_run_timestamp()
    python_task(run_timestamp)
    # 其他任务也可以直接复用run_timestamp变量
    # another_task(run_timestamp)

my_dag()

内容的提问来源于stack exchange，提问作者Matheus Oliveira