Airflow中在DAG代码块内访问params参数报错，如何解决？

为什么会报错？ #

你使用的{{ params.proj }}是Airflow的模板语法，这种语法仅在Airflow任务的模板字段（比如PythonOperator的op_args/op_kwargs、BashOperator的bash_command等）中才会被渲染解析。而你直接在DAG定义的Python代码里调用test()函数，这部分代码在Airflow加载DAG时就会被执行，此时模板语法不会被处理，Python解释器会把{{ params.proj }}当成普通代码，自然找不到params变量，因此抛出NameError。

解决方案分两种场景： #

场景1：仅在DAG加载阶段访问params #

如果只是想在DAG加载时（比如初始化变量、调用普通Python函数）使用DAG定义的params，直接从dag对象的params属性中取值即可：

修正后的代码：

from datetime import datetime, timedelta
from airflow import DAG

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

with DAG(
        dag_id="start-matrix",
        default_args=default_args,
        render_template_as_native_obj=True,
        params={
            'proj' : "team",
            'username' : "leebee",
        },
        schedule_interval=None,
        start_date=datetime(2021, 4, 5, 15, 0),
        catchup=False
) as dag:
    def test(proj, name, dag):
        print(proj + " " + name)

    # 直接从dag.params中取值
    test(dag.params['proj'], dag.params['username'], dag)

场景2：在Airflow任务中使用params #

如果你的目的是把test()作为Airflow的任务执行，需要使用PythonOperator（或其他对应Operator），此时可以通过context或模板语法获取params：

示例代码：

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

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

def test(**context):
    # 通过context获取params
    proj = context['params']['proj']
    username = context['params']['username']
    print(proj + " " + username)

with DAG(
        dag_id="start-matrix",
        default_args=default_args,
        render_template_as_native_obj=True,
        params={
            'proj' : "team",
            'username' : "leebee",
        },
        schedule_interval=None,
        start_date=datetime(2021, 4, 5, 15, 0),
        catchup=False
) as dag:
    # 定义PythonOperator任务
    test_task = PythonOperator(
        task_id='test_task',
        python_callable=test,
        # 也可以通过模板语法传递params到op_kwargs
        # op_kwargs={
        #     'proj': '{{ params.proj }}',
        #     'username': '{{ params.username }}'
        # },
        provide_context=True  # 开启后，context会被传入python_callable
    )

总结 #

• DAG加载阶段：直接用dag.params访问定义的参数
• 任务执行阶段：通过context或模板语法获取params，前提是使用Airflow的Operator定义任务

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