Airflow DAG传递数组/整数参数遇Jinja2模板语法错误求助

错误根源 #

你混淆了DAG解析阶段的Python代码逻辑和任务运行阶段的Jinja模板渲染：

• {{ params.sourcedir }} 是Jinja模板语法，仅能在任务的模板化字段（如op_args、bash_command）中使用，用于任务运行时动态读取参数。
• 你直接在DAG定义的Python循环（range(0,len('{{ params.sourcedir}}'))）中使用Jinja字符串，此时Jinja尚未渲染，'{{ params.sourcedir}}'只是普通字符串，既不是数组也无法计算长度，自然触发语法错误。
• 另外原代码存在语法错误：range(0,len('{{ params.sourcedir}}'))) 多了一个右括号；循环生成的task_id重复为taskinfo，Airflow会拒绝重复的任务ID。

分场景解决方案 #

场景1：参数是固定值（DAG解析时已知） #

如果数组/整数参数是写死在DAG定义里的，直接用Python变量存储，无需Jinja模板：

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

default_args = {
    'owner': 'airflow',
}

# 先定义Python变量，DAG解析阶段可直接调用
source_dirs = ['/home/arya/']
timenum = 0

with DAG(
    dag_id="INSURANCE_CALC",
    start_date=datetime(2022, 1, 24),
    schedule_interval=None,
    default_args=default_args,
    params={
        "param2": "arya2@gmail.com",
        "sourcedir": source_dirs,
        "timenum": timenum
    },
    catchup=False
) as dag:

    # 直接遍历Python数组变量，生成唯一任务ID
    for idx, dir_path in enumerate(source_dirs):
        modified_dir = f"abc{dir_path}xyz"
        taskinfo = PythonOperator(
            task_id=f"taskinfo_{idx}",
            python_callable=lambda dir: print(f"处理目录: {dir}"),
            op_kwargs={"dir": modified_dir}
        )

场景2：参数是动态传入（运行时才确定） #

如果数组/整数参数需要在DAG手动触发时动态修改，使用Airflow 2.3+支持的动态任务映射功能，在任务运行时解析Jinja参数：

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

default_args = {
    'owner': 'airflow',
}

# 任务处理逻辑
def process_directory(dir_path):
    modified_dir = f"abc{dir_path}xyz"
    print(f"处理目录: {modified_dir}")
    # 这里添加你的业务代码

with DAG(
    dag_id="INSURANCE_CALC",
    start_date=datetime(2022, 1, 24),
    schedule_interval=None,
    default_args=default_args,
    params={
        "param2": "arya2@gmail.com",
        "sourcedir": ['/home/arya/'],  # 触发DAG时可修改此参数
        "timenum": 0
    },
    catchup=False
) as dag:

    # 基于params.sourcedir动态生成任务
    taskinfo = PythonOperator.partial(
        task_id="taskinfo",
        python_callable=process_directory
    ).expand(
        op_args="{{ params.sourcedir }}"  # Jinja模板在这里生效，自动解析数组
    )

关键注意事项 #

1. Jinja模板的适用范围：仅能在Operator支持的模板化字段中使用，不能在DAG定义的Python逻辑（如循环、变量赋值）中直接调用。
2. 任务ID唯一性：循环生成任务时，必须保证每个task_id唯一，否则Airflow会抛出任务重复的错误。
3. 动态任务映射版本要求：expand/partial语法需要Airflow 2.3及以上版本支持，若使用旧版本，可在PythonOperator的python_callable中直接读取params并处理数组。

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