Apache Airflow任务中应使用什么替代now()函数？

当你需要在Airflow任务中生成固定的loadtimestamp（而非每次运行都变化的当前时间），可以用以下几种Airflow原生的时间变量来替代datetime.now()，这些值在任务的整个调度周期内是固定的，不会因重试、延迟执行改变：

1. 使用execution_date（传统调度场景首选） #

execution_date是Airflow经典调度模型的核心时间变量，代表当前任务所属调度周期的执行时间。比如你的DAG按小时调度，execution_date就是该小时的整点时间（如2024-05-20T10:00:00+00:00），无论任务是准时运行还是延迟/重试，这个值都不会变，非常适合作为loadtimestamp标记数据的所属周期。

代码示例： #

• 通过PythonOperator的context获取：

from airflow.decorators import task

@task
def load_data(**context):
    execution_dt = context["execution_date"]
    load_timestamp = execution_dt.strftime("%Y-%m-%d %H:%M:%S")
    # 用load_timestamp执行数据插入等关键操作
    print(f"Load timestamp: {load_timestamp}")

• 通过Jinja模板直接传入：

from airflow.operators.python import PythonOperator

def load_data(load_timestamp):
    # 使用传入的load_timestamp执行操作
    print(f"Load timestamp: {load_timestamp}")

PythonOperator(
    task_id="load_data_task",
    python_callable=load_data,
    op_kwargs={"load_timestamp": "{{ execution_date.strftime('%Y-%m-%d %H:%M:%S') }}"}
)

2. 使用data_interval_start/data_interval_end（Airflow 2.2+推荐） #

如果你使用Airflow 2.2及以上版本，且采用了Timetable自定义调度逻辑，推荐使用data_interval_start和data_interval_end。这两个变量代表当前任务处理的数据区间起止时间，比execution_date更贴合现代Airflow的调度语义。比如按天调度的DAG，data_interval_start是当天0点，通常用它作为loadtimestamp即可。

代码示例： #

from airflow.decorators import task

@task
def load_data(**context):
    data_interval_start = context["data_interval_start"]
    load_timestamp = data_interval_start.strftime("%Y-%m-%d %H:%M:%S")
    # 执行数据插入等操作

3. 使用task_instance.start_date（任务首次启动时间） #

如果你的loadtimestamp需要标记任务实际首次启动的时间（而非调度周期时间），可以用task_instance.start_date。这个值是任务第一次开始运行的时间，即使任务重试，该值也不会更新，适合需要记录任务首次触发时间的场景。

代码示例： #

from airflow.decorators import task

@task
def load_data(**context):
    ti = context["task_instance"]
    load_timestamp = ti.start_date.strftime("%Y-%m-%d %H:%M:%S")
    # 执行数据操作

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