Airflow使用Jinja宏作为变量创建动态DAG报错求解

你遇到的报错核心原因是：Jinja模板语法（被{{}}包裹的内容）仅会在任务运行时才会被渲染为实际值，在DAG文件解析阶段，模板内容只是普通字符串。你当前的写法在DAG解析阶段就调用get_dag_daterange函数计算日期范围、生成任务，此时start_dag_params和end_dag_params要么是字符串类型的模板，要么是变量里读出来的日期字符串，无法直接做日期运算，因此触发类型错误。

另外你依赖的next_execution_date属于运行时上下文参数，DAG解析阶段根本无法拿到该值，因此你无法在解析阶段就生成固定的任务列表，必须使用Airflow 2.3及以上版本提供的动态任务映射功能，在任务运行时根据实际渲染后的日期生成子任务。

from airflow import DAG
from airflow.models import Variable
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.decorators import task
from datetime import timedelta
import pendulum

# 通用默认参数可自行补充
default_args = {
    "owner": "airflow",
    "start_date": pendulum.datetime(2024,1,1,tz="Asia/Shanghai")
}

dag_vars = Variable.get("dag_dates", deserialize_json=True) 
DAG_NAME = "dag_test"

@task
def get_date_range(**context):
    # 运行时获取渲染后的起止日期
    if dag_vars["dag_start_dt"] == "NULL":
        start_dt = context["next_execution_date"] - timedelta(days=1)
    else:
        start_dt = pendulum.parse(dag_vars["dag_start_dt"])
    if dag_vars["dag_end_dt"] == "NULL":
        end_dt = context["next_execution_date"]
    else:
        end_dt = pendulum.parse(dag_vars["dag_end_dt"])
    # 生成日期列表返回，用于后续动态映射
    date_list = []
    for n in range(int((end_dt - start_dt).days)):
        date_list.append((start_dt + timedelta(n)).strftime("%Y-%m-%d"))
    return date_list

dag = DAG(
    dag_id=DAG_NAME,
    default_args=default_args,
    schedule_interval=None,
    concurrency=1,
    max_active_runs=1,
    catchup=False
)

with dag:
    date_list = get_date_range()
    # 动态映射生成多个Postgres任务
    PostgresOperator.partial(
        sql = "CALL test_procedure('{{ params.one_date }}')",
        postgres_conn_id = "xxx",
        pool = "pool_test",
        autocommit = True
    ).expand(
        task_id = date_list.map(lambda x: f"test_procedure_{x}"),
        params = date_list.map(lambda x: {"one_date": x})
    )

• 把日期范围计算逻辑放到带@task装饰器的Python函数中，该函数会在任务运行时执行，此时可以通过context参数拿到next_execution_date等运行时上下文变量，同时可以直接做日期运算
• 用PostgresOperator.partial定义任务通用参数，再通过expand方法根据返回的日期列表动态生成多个子任务，每个子任务对应一个日期的存储过程调用
• 所有运行时才确定的参数全部交由Airflow内置的渲染逻辑处理，不需要在DAG解析阶段手动处理

内容的提问来源于stack exchange，提问作者ann.shrudt