Airflow如何将输入变量用作默认参数且变量缺失时安全回退

Airflow 原生支持「将运行时传入的输入变量作为default_args取值，未传参时自动回退到预设替代值」的能力，你之前的实现报错是写法不符合Airflow的参数加载、Jinja渲染逻辑导致的，不是功能不支持。

你写的Jinja判断逻辑存在两个问题：

• 存在性判断逻辑失效：Airflow在任务渲染前会将DAG定义的默认params、default_args参数、触发DAG时传入的运行时参数做三层合并，如果你没有提前声明example参数键，直接在模板里用params['example']做键访问，会在你的if 'example' in params判断逻辑生效前，就触发Jinja的未定义变量错误。部分Airflow版本中params是代理对象，'example' in params的判断会误返回True，直接走到取值分支抛错。
• 取值方式不安全：直接用方括号[]访问字典键本身就没有兜底逻辑，键不存在时必然抛异常。

方案1：使用Airflow原生params默认值机制（官方推荐，最稳定） #

不要在default_args里写Jinja模板做参数判断，直接在DAG初始化时提前声明params的默认值，Airflow会自动按优先级合并参数：运行时传入的参数 > 任务级params > DAG级params > default_args参数，未传参时自动使用你预设的兜底值。
示例代码：

from airflow import DAG
from airflow.decorators import task
from datetime import datetime

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

with DAG(
    dag_id="param_fallback_demo",
    default_args=default_args,
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    # 提前声明所有参数的兜底默认值
    params={
        "example": None
    }
) as dag:
    @task
    def process_param(**context):
        # 直接取值即可，未传参时自动取预设的None，传参则使用传入值
        example_value = context["params"]["example"]
        print(f"当前example参数取值：{example_value}")

    process_param()

这个方案不需要写任何Jinja判断逻辑，参数合并全由Airflow内置逻辑完成，不会出现渲染顺序导致的异常，也是生产环境最常用的写法。

方案2：在default_args中用安全取值写法写Jinja模板 #

如果你确实需要把参数兜底逻辑写在default_args里，不要用方括号按键取值，改用字典的get()方法做安全取值，或者用Jinja自带的default过滤器，从根源避免键不存在的异常。
示例代码：

default_args = {
    # 字典get方法安全取值，键不存在时直接返回指定的None作为兜底
    "example": "{{ params.get('example', None) }}",
    # 等价的default过滤器写法，注意要加boolean=True确保空值也能触发兜底
    # "example": "{{ params.example | default(None, boolean=True) }}"
}

这种写法不需要提前在DAG params里声明键，运行时传了example就用传入值，没传就自动返回None，不会抛UndefinedError。

• 不要在default_args里写过于复杂的Jinja逻辑，参数兜底优先用方案1的原生params机制，可维护性更高，也不会因为Airflow版本升级导致模板渲染逻辑失效。
• Airflow 2.3及以上版本支持在DAG中定义参数表单，可以直接在UI上配置参数的类型、默认值、校验规则，比写在default_args里更直观。

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