如何在Airflow中打印时间并以打印语句输出到日志

直接在任务逻辑里读取Airflow注入的上下文时间变量，用print或者标准logging模块输出，内容会自动被采集到任务日志中，不需要额外做日志配置。

常用可打印的Airflow内置时间变量 #

• execution_date：任务对应的逻辑调度时间，是DAG配置的计划运行时间，不是任务实际启动的物理时间，是做增量数据处理最常用的时间参数
• prev_execution_date：当前任务上一个调度周期的逻辑执行时间
• next_execution_date：当前任务下一个调度周期的逻辑执行时间
• dag_run.start_date：本次DAG运行被实际触发的物理时间
• task_instance.start_date：当前任务实例实际开始执行的物理时间

代码示例 #

传统Operator写法（兼容1.x/2.x版本） #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def output_airflow_time(**context):
    print("="*20 + " Airflow时间信息 " + "="*20)
    print(f"逻辑调度时间(execution_date): {context['execution_date']}")
    print(f"上一周期调度时间: {context['prev_execution_date']}")
    print(f"下一周期调度时间: {context['next_execution_date']}")
    print(f"DAG实际触发时间: {context['dag_run'].start_date}")
    print(f"当前任务启动时间: {context['task_instance'].start_date}")
    # 需要格式化时间字符串可以直接调用strftime转换
    print(f"格式化逻辑时间(非东八区配置注意自行转换时区): {context['execution_date'].strftime('%Y-%m-%d %H:%M:%S')}")

with DAG(
    dag_id="print_time_demo",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:
    print_time_task = PythonOperator(
        task_id="print_time_task",
        python_callable=output_airflow_time,
        # Airflow 1.x版本需要配置该参数开启上下文传入，2.x版本默认开启可省略
        provide_context=True
    )

TaskFlow API写法（Airflow 2.x推荐） #

直接把需要的时间变量作为任务函数的参数声明即可，不需要手动读取context：

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

@dag(start_date=datetime(2024, 1, 1), schedule="@daily", catchup=False)
def airflow_time_print_demo():
    @task
    def print_time(
        execution_date=None,
        prev_execution_date=None,
        next_execution_date=None,
        dag_run=None
    ):
        print(f"逻辑调度时间: {execution_date}")
        print(f"DAG实际触发时间: {dag_run.start_date}")

    print_time()

airflow_time_print_dag()

注意：如果打印的时间和你预期的本地时间不一致，先检查Airflow配置的默认时区，默认配置是UTC时间，需要国内时间的话可以在airflow.cfg里把default_timezone配置为Asia/Shanghai。

任务运行后，直接在Airflow UI打开对应任务实例的日志页面，就能看到所有你打印的时间字段了。

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