Airflow 2.x DAG报错：未使用Task Group却触发AttributeError

核心原因 #

该错误并非由显式使用Task Group导致，而是Airflow 2.x与1.10.x的DAG/任务定义语法不兼容，导致代码中某个函数被错误识别为DAG对象，触发内部逻辑尝试访问不存在的task_group属性。常见场景如下：

1. DAG装饰器误用 #

Airflow 2.x的@dag装饰器会将函数包装为DAG实例，若你沿用1.10.x的DAG类实例化写法，同时错误叠加@dag装饰器，会导致函数与DAG对象的引用冲突。例如：

# 错误示例：混用新旧DAG定义方式
from airflow.decorators import dag
from airflow.models import DAG
from datetime import datetime

def FILE_DB_PRODUCT_TABLES():
    default_args = {'owner': 'airflow', 'start_date': datetime(2024,1,1)}
    dag = DAG(dag_id='FILE_DB_PRODUCT_TABLES', default_args=default_args)
    # 任务定义...
    return dag

# 此处错误叠加装饰器，将函数转换为DAG实例，内部逻辑混乱
FILE_DB_PRODUCT_TABLES = dag()(FILE_DB_PRODUCT_TABLES)

2. 任务与DAG关联错误 #

Airflow 2.x中@task装饰的任务需正确绑定到DAG上下文。若你在@dag装饰的函数外部定义任务，或未显式指定所属DAG，任务会尝试从错误的对象（如普通函数）中获取task_group属性，触发报错。例如：

# 错误示例：任务未关联到DAG上下文
from airflow.decorators import task
from airflow.models import DAG
from datetime import datetime

@task
def extract():
    pass

def FILE_DB_PRODUCT_TABLES():
    default_args = {'owner': 'airflow', 'start_date': datetime(2024,1,1)}
    with DAG('FILE_DB_PRODUCT_TABLES', default_args=default_args) as dag:
        # 任务未绑定到当前DAG，内部逻辑出错
        extract_task = extract()

3. 环境依赖冲突 #

若环境中同时残留Airflow 1.x的依赖文件，或导入了版本不匹配的模块（如混用airflow.models.DAG与airflow.decorators.dag），会导致DAG对象结构异常，出现属性缺失。

排查与修复步骤 #

1. 统一DAG定义语法：改用Airflow 2.x标准的@dag装饰器写法，示例：

from airflow.decorators import dag, task
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1)
}

@dag(dag_id='FILE_DB_PRODUCT_TABLES', default_args=default_args, schedule_interval='@daily')
def file_db_product_tables_dag():
    @task
    def extract():
        return "raw_data"
    
    @task
    def transform(data):
        return data.upper()
    
    @task
    def load(data):
        print(f"Loaded: {data}")
    
    # 定义任务依赖
    extract() >> transform() >> load()

# 实例化DAG
file_db_product_tables_dag = file_db_product_tables_dag()

2. 确保任务在DAG上下文内定义：所有@task装饰的函数必须放在@dag装饰的函数内部，或通过task(dag=dag)显式指定所属DAG。
3. 验证Airflow版本：执行airflow version确认环境为纯2.x版本，无旧版本依赖残留。
4. 定位报错代码行：查看报错栈中触发错误的具体代码位置（如某行任务实例化代码），确认是否存在对象引用混淆的情况。

额外提示 #

Airflow 2.x已移除1.x的set_upstream/set_downstream等API，统一使用>>/<<操作符定义任务依赖，避免因API不兼容导致的隐式错误。

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