Airflow调度执行时运行时输入变量默认值配置失败求助

问题根源 #

调度触发的DAG实例（dag_run）默认没有conf属性值，会返回None；而手动触发时无论是否输入参数，dag_run.conf都会是一个字典（空字典或含输入参数的字典）。你之前的代码直接调用kwargs['dag_run'].conf.get()，调度时conf为None就会触发报错。

另外你配置default_dag_args里的params未生效，是因为代码里没有从params取值，而是依然从dag_run.conf获取，两者是Airflow中不同的参数传递渠道。

正确解决方案 #

方案1：兼容dag_run.conf为None的场景 #

修改取值代码，先判断dag_run和conf是否存在，再执行get操作：

def branch_func(**kwargs):
    # 先判断dag_run是否存在，再判断conf是否不为None
    if kwargs.get('dag_run') and kwargs['dag_run'].conf:
        cleanup = kwargs['dag_run'].conf.get('cleanup', 'N')
    else:
        cleanup = 'N'
    # 后续分支逻辑...

也可以用更简洁的链式判断写法：

cleanup = kwargs['dag_run'].conf.get('cleanup', 'N') if (kwargs.get('dag_run') and kwargs['dag_run'].conf) else 'N'

方案2：改用Airflow官方推荐的params参数（更规范） #

这是Airflow设计用来传递DAG参数的标准方式，调度和手动触发都能自动继承默认值，手动触发时也能在UI中覆盖参数：

1. 在default_dag_args中配置默认参数：

default_dag_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    # 配置默认参数
    'params': {
        'cleanup': 'N'
    }
}

2. 修改BranchPythonOperator的取值代码，从kwargs['params']获取：

def branch_func(**kwargs):
    # 直接从params取，调度时用默认值，手动触发时用传入值
    cleanup = kwargs['params'].get('cleanup', 'N')
    # 后续分支逻辑...

手动触发时，在Airflow UI的"Trigger DAG w/ config"界面，你可以在Params区域输入{"cleanup": "Y"}来覆盖默认值，和之前用conf的效果一致，但调度时不会出现None的问题。

方案3：临时方案优化（替代try-except） #

如果暂时不想改取值逻辑，也可以用get方法链式取值，避免直接访问属性报错：

cleanup = kwargs.get('dag_run', {}).get('conf', {}).get('cleanup', 'N')

这里通过多次调用get，当某一层为None时会返回空字典，后续get不会报错，直接返回默认值'N'。

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