Airflow报错AttributeError：'function'对象无'task_group'属性求助

核心原因 #

AttributeError: 'function' object has no attribute 'task_group' 本质是你误将普通函数当作Airflow的DAG或TaskGroup对象调用task_group属性。常见触发场景：

• DAG工厂函数返回的是未执行的函数引用，而非初始化完成的DAG实例
• 调用.task_group()的对象被错误赋值为函数，而非Airflow核心对象
• 导入或赋值时混淆了函数与类（比如把创建DAG的函数当作DAG类使用）

排查修复步骤 #

1. 校验DAG工厂函数返回值 #

确保工厂函数最终返回DAG实例，而非函数本身：

# 错误示例：返回函数而非DAG实例
def create_dag(mongo_data):
    def build_dag():
        with DAG(dag_id=mongo_data['dag_id']) as dag:
            pass
        return dag
    return build_dag  # 返回的是函数，不是实例

# 正确示例：直接返回DAG实例
def create_dag(mongo_data):
    with DAG(
        dag_id=mongo_data['dag_id'],
        schedule=mongo_data['schedule']
    ) as dag:
        # 任务/TaskGroup定义
    return dag

2. 检查task_group调用对象 #

确认调用.task_group()的是DAG或TaskGroup实例，而非函数：

# 错误示例：把工厂函数当作DAG对象
dag_builder = create_dag(mongo_data)
dag_builder.task_group(group_id='my_group')  # dag_builder是函数，不是实例

# 正确示例：先获取DAG实例再操作
dag = create_dag(mongo_data)
with dag.task_group(group_id='my_group') as tg:
    # 子任务定义

3. 验证MongoDB数据传递逻辑 #

确保从MongoDB获取的参数是合法类型，没有混入函数引用：

• 确认dag_id、schedule等字段为字符串/合法值，而非函数
• 传递数据时避免将未执行的函数对象传入工厂函数

4. 强制触发Airflow重新解析 #

修改代码后，重启Airflow调度器/网页服务，或删除$AIRFLOW_HOME/dags下的临时解析文件，强制DAG重新解析。

额外排查技巧 #

在DAG工厂函数中添加日志，输出关键对象类型：

import logging

def create_dag(mongo_data):
    logging.info(f"MongoDB传入数据: {mongo_data}")
    with DAG(...) as dag:
        logging.info(f"DAG实例类型: {type(dag)}")
        # 任务定义
    return dag

查看Airflow调度器日志，确认对象类型是否符合预期。

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