Airflow SubDag内访问context报错ds不存在问题咨询

核心报错原因 #

你遇到的KeyError: 'ds'和SubDag本身能不能访问上下文没有关系，问题出在函数的执行时机和上下文的生效阶段不匹配：

• create_sensor_subdag 是DAG解析阶段就会被直接调用执行的函数：你在定义SubDagOperator的时候直接将create_sensor_subdag(...)的返回值赋值给了subdag参数，此时Airflow还处于扫描加载DAG文件的阶段，没有任何任务在运行，自然不存在运行时上下文，context字典是空的，取ds自然会抛出异常，这也是提示Broken DAG的原因——DAG在解析阶段就失败了，根本没有到任务运行的环节。
• 你设置的provide_context=True 作用对象是SubDag内部的运行时任务，不会把上下文传递给用来创建SubDag的create_sensor_subdag函数，两个参数的生效范围完全不同。

SubDag访问上下文的正确方式 #

SubDag内部的任务在运行阶段是可以正常访问上下文的，只是不能在创建SubDag结构的函数里直接获取。如果你需要在SubDag的任务中使用ds这类上下文变量，有两种常见实现方式：

1. 直接在任务的可模板化参数中用Jinja模板引用，Airflow运行时会自动渲染变量值，示例：

# 在create_sensor_subdag中定义Sensor任务时直接写模板即可
FileSensor(
    task_id="check_file",
    filepath=f"/data/{{{{ ds }}}}/target_file.txt",
    dag=sub_dag
)

2. 给SubDag内的任务设置provide_context=True，在任务的执行函数里取上下文：

def sensor_exec_logic(**context):
    execution_date = context["ds"]
    # 执行具体的检测逻辑

PythonOperator(
    task_id="sensor_task",
    python_callable=sensor_exec_logic,
    provide_context=True,
    dag=sub_dag
)

额外提示：Airflow 2.x 开始已经不推荐使用SubDagOperator，更建议用TaskGroup实现任务分组，没有SubDag的上下文、调度权限等各类坑，使用成本更低。

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