如何在Airflow DAG中捕获统一时间变量，实现多表同时间戳插入？

你当前代码的问题在于，start_task_time = datetime.now(timezone.utc)是在Airflow调度器解析DAG文件时执行的，而Airflow会定期重新解析DAG，且两个任务的SQL语句通过f-string渲染的时机可能存在细微差异，导致最终插入的时间戳不一致。此外，这种写法混淆了DAG解析时间和任务实际执行时间，不符合Airflow的运行逻辑。

解决方案 #

方案一：使用Airflow内置上下文变量（推荐） #

利用Airflow提供的DAG运行上下文变量（如data_interval_start或execution_date），同一个DAG运行实例（DAG Run）中所有任务的该变量值完全一致，适合标记批量任务的逻辑时间。

修改后的代码：

from airflow.utils.dates import days_ago
from airflow.providers.postgres.operators.postgres import PostgresOperator

with DAG("inform_status",
         schedule_interval="55 17 * * 1-5",
         start_date=days_ago(1),
         catchup=False,
         tags=["Inform Status"]) as dag:

create_report_1 = PostgresOperator(
         task_id='create_report_1',
         postgres_conn_id='inform_status',
         sql="""
            INSERT into table1 (created_date, dbid)
            SELECT '{{ data_interval_start.astimezone(utc) }}'::timestamp with time zone, dbid
            FROM pg_catalog.gp_segment_configuration;
         """
         )
create_report_2 = PostgresOperator(
         task_id='create_report_2',
         postgres_conn_id='inform_status',
         sql="""
            INSERT into table2 (created_date, dbid)
            SELECT '{{ data_interval_start.astimezone(utc) }}'::timestamp with time zone, dbid
            FROM pg_catalog.gp_segment_configuration;
         """
         )

• 说明：data_interval_start是当前DAG运行的时间区间起始时间，与调度规则严格对应；若需要任务实际触发的时间，可替换为{{ execution_date.astimezone(utc) }}。PostgresOperator支持Jinja2模板语法，会在任务执行时自动渲染变量。

方案二：通过PythonOperator生成统一时间戳并传递 #

如果需要任务实际执行时刻的时间戳，可先通过PythonOperator生成一个统一时间，再通过XCom传递给后续的PostgresOperator，确保所有任务使用同一时间戳。

修改后的代码：

from airflow.utils.dates import days_ago
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.operators.python import PythonOperator
from datetime import datetime, timezone

def generate_task_time(**context):
    # 生成UTC标准格式的时间戳
    task_time = datetime.now(timezone.utc).isoformat()
    # 将时间戳推送到XCom供其他任务获取
    context['ti'].xcom_push(key='task_time', value=task_time)

with DAG("inform_status",
         schedule_interval="55 17 * * 1-5",
         start_date=days_ago(1),
         catchup=False,
         tags=["Inform Status"]) as dag:

# 先执行生成时间戳的任务
get_task_time = PythonOperator(
    task_id='get_task_time',
    python_callable=generate_task_time,
    provide_context=True
)

create_report_1 = PostgresOperator(
         task_id='create_report_1',
         postgres_conn_id='inform_status',
         sql="""
            INSERT into table1 (created_date, dbid)
            SELECT '{{ ti.xcom_pull(task_ids="get_task_time", key="task_time") }}'::timestamp with time zone, dbid
            FROM pg_catalog.gp_segment_configuration;
         """
         )
create_report_2 = PostgresOperator(
         task_id='create_report_2',
         postgres_conn_id='inform_status',
         sql="""
            INSERT into table2 (created_date, dbid)
            SELECT '{{ ti.xcom_pull(task_ids="get_task_time", key="task_time") }}'::timestamp with time zone, dbid
            FROM pg_catalog.gp_segment_configuration;
         """
         )

# 设置任务依赖：先生成时间戳，再执行插入任务
get_task_time >> [create_report_1, create_report_2]

方案三：使用数据库端当前时间函数（简易但有局限） #

如果可以接受两个任务的执行时间差极小，可直接使用PostgreSQL的内置时间函数CURRENT_TIMESTAMP或NOW()，但该时间是数据库执行SQL时的时间，若两个任务串行执行，时间戳会存在差异。

修改后的代码：

from airflow.utils.dates import days_ago
from airflow.providers.postgres.operators.postgres import PostgresOperator

with DAG("inform_status",
         schedule_interval="55 17 * * 1-5",
         start_date=days_ago(1),
         catchup=False,
         tags=["Inform Status"]) as dag:

create_report_1 = PostgresOperator(
         task_id='create_report_1',
         postgres_conn_id='inform_status',
         sql="""
            INSERT into table1 (created_date, dbid)
            SELECT CURRENT_TIMESTAMP, dbid
            FROM pg_catalog.gp_segment_configuration;
         """
         )
create_report_2 = PostgresOperator(
         task_id='create_report_2',
         postgres_conn_id='inform_status',
         sql="""
            INSERT into table2 (created_date, dbid)
            SELECT CURRENT_TIMESTAMP, dbid
            FROM pg_catalog.gp_segment_configuration;
         """
         )

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