Apache Airflow：传递DAG对象创建自定义TaskGroup报错求助

问题场景 #

尝试自定义TaskGroup类替换Airflow中的SubDAG，希望直接传递DAG对象而非使用上下文管理器，以减少现有DAG的代码改动。运行代码后触发报错：

airflow.exceptions.AirflowException: Tried to create relationships between tasks that don't have DAGs yet. Set the DAG for at least one task and try again: [<Task(_PythonDecoratedOperator): TestTaskGroup.emptyTask>, <Task(_PythonDecoratedOperator): TestTaskGroup.pythonTask>]

自定义TaskGroup代码：

from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
from airflow.utils.task_group import TaskGroup
from airflow.decorators import task

class TestTaskGroup(TaskGroup):
    def __init__(self, dag,  group_id="TestTaskGroup", *args, **kwargs):
        super().__init__(dag=dag,group_id=group_id, *args, **kwargs)
        self.dag = dag

        @task(task_group=self)
        def emptyTask():            
            EmptyOperator(task_id="emptytask", dag=self.dag)
            

        @task(task_group=self)
        def pythonTask():            
            PythonOperator(task_id="pythontask", dag=self.dag, python_callable=lambda:print("Hello World"))

        emptyTask() >> pythonTask()

testdag.py代码：

from taskgroups.testtaskgroup import TestTaskGroup
from airflow import DAG
from datetime import datetime
from airflow.operators.empty import EmptyOperator

dag = DAG(
    dag_id="testdag",
    start_date = datetime(2024,2,26),
    schedule_interval="@daily",
    catchup=False,
    max_active_runs=1,
    default_args={
        "owner":"hemant.sah",
    }
) 

sometask = EmptyOperator(task_id="sometask", dag=dag)
grouptask = TestTaskGroup(dag=dag)

sometask >> grouptask

错误原因 #

1. @task装饰器用法错误：@task装饰器的作用是将Python函数封装为_PythonDecoratedOperator，函数体应直接编写业务逻辑，而非在内部手动创建Operator实例，这种写法会导致装饰器生成的任务与手动创建的Operator出现双重关联冲突。
2. DAG关联冗余混乱：手动给内部Operator指定dag=self.dag，同时又通过task_group=self关联到TaskGroup，导致任务的DAG归属冲突。实际上TaskGroup已关联DAG，内部任务会自动继承该DAG，无需手动指定。

解决方案 #

修改自定义TaskGroup的实现，正确使用@task装饰器或直接创建Operator并关联到TaskGroup，确保任务通过TaskGroup自动继承DAG：

修改后的自定义TaskGroup代码 #

from airflow.operators.empty import EmptyOperator
from airflow.utils.task_group import TaskGroup
from airflow.decorators import task

class TestTaskGroup(TaskGroup):
    def __init__(self, dag, group_id="TestTaskGroup", *args, **kwargs):
        super().__init__(dag=dag, group_id=group_id, *args, **kwargs)
        
        # 方式1：直接创建EmptyOperator，指定task_group为当前TaskGroup
        empty_task = EmptyOperator(task_id="emptytask", task_group=self)
        
        # 方式2：正确使用@task装饰器，函数体编写业务逻辑，无需手动创建PythonOperator
        @task(task_group=self)
        def python_task():
            print("Hello World")
        
        # 建立任务依赖
        empty_task >> python_task()

testdag.py代码（无需改动，保持原逻辑即可） #

from taskgroups.testtaskgroup import TestTaskGroup
from airflow import DAG
from datetime import datetime
from airflow.operators.empty import EmptyOperator

dag = DAG(
    dag_id="testdag",
    start_date = datetime(2024,2,26),
    schedule_interval="@daily",
    catchup=False,
    max_active_runs=1,
    default_args={
        "owner":"hemant.sah",
    }
) 

sometask = EmptyOperator(task_id="sometask", dag=dag)
grouptask = TestTaskGroup(dag=dag)

sometask >> grouptask

关键修改点 #

• 移除@task装饰器函数内部手动创建Operator的错误写法，让装饰器直接封装业务逻辑。
• 传统Operator（如EmptyOperator）直接创建时，通过task_group=self关联到当前TaskGroup，无需指定dag参数，TaskGroup已关联的DAG会自动被内部任务继承。
• 确保任务依赖建立时，所有任务都已通过TaskGroup正确关联到DAG，避免出现无DAG的任务关联操作。

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