Airflow任务实例中trigger_date_time值持续自动更新问题求助

Airflow默认每分钟会重新解析一次DAG文件，你在DAG定义的顶层直接执行datetime.now()，每次解析DAG时这段代码都会重新运行，所以trigger_date_time的值会不断更新，没法固定在DAG触发执行的时间点。

要让trigger_date_time固定为DAG触发时的时间，得用Airflow任务上下文里的内置时间变量——这些变量在DAG触发时就已经确定，且每个任务实例都有独立的取值：

• Python任务实现方式
  在Python任务函数中，通过context参数获取execution_date（或Airflow 2.x中的data_interval_start），再格式化为你需要的字符串格式：

  from airflow.decorators import task
  from airflow import DAG
  from datetime import datetime, timezone
  
  with DAG(
      dag_id="your_dag_id",
      start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
      schedule_interval="@daily"
  ) as dag:
  
      @task
      def process_data(**context):
          # 将执行时间格式化为目标字符串
          trigger_date_time = context["execution_date"].strftime("%Y-%m-%d_%H:%M:%S%z").replace("+", "_")
          print(f"固定触发时间: {trigger_date_time}")
          # 这里写你的业务逻辑
  
      process_data()
  

• 非Python任务（如BashOperator）实现方式
  直接用Airflow的模板语法在任务参数中引用内置变量：

  from airflow.operators.bash import BashOperator
  from airflow import DAG
  from datetime import datetime, timezone
  
  with DAG(
      dag_id="your_dag_id",
      start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
      schedule_interval="@daily"
  ) as dag:
  
      bash_task = BashOperator(
          task_id="bash_task",
          bash_command='echo "固定触发时间: {{ execution_date.strftime("%Y-%m-%d_%H:%M:%S%z").replace("+", "_") }}"'
      )
  

• 别在DAG定义的顶层写动态计算的代码（比如直接调用datetime.now()），因为DAG解析过程会反复执行这段代码，导致值不稳定。
• execution_date是Airflow的经典触发时间变量，Airflow 2.x中也可以用data_interval_start，多数调度场景下两者取值一致，可根据需求选择。

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