Airflow：基于装饰器语法配置TaskGroup与外部任务的依赖

错误原因 #

你代码中被@task_group装饰的tg1函数没有返回任何值，调用tg1()时得到的是None。当执行start >> tg1() >> end时，Airflow无法对None对象执行依赖关系更新操作，因此抛出该错误。

解决方案 #

以下两种方式可基于TaskFlow API装饰器语法实现TaskGroup与外部任务的依赖配置：

方式1：将TaskGroup实例赋值给变量后配置依赖 #

把tg1()的调用结果存储为变量，通过变量建立外部依赖：

from datetime import datetime

from airflow.decorators import dag, task_group
from airflow.operators.dummy import DummyOperator


@dag(
    start_date=datetime(2021, 4, 20, 15, 0),
    concurrency=5,
    max_active_runs=1,
    catchup=False,
    schedule_interval="0 17 * * *"
)
def sv_load():
    start = DummyOperator(task_id="start")
    end = DummyOperator(task_id="end")

    @task_group(group_id="tg_test")
    def tg1():
        t1 = DummyOperator(task_id="task1")
        t2 = DummyOperator(task_id="task2")
        t1 >> t2

    # 保存TaskGroup实例到变量
    tg_instance = tg1()
    start >> tg_instance >> end

dag = sv_load()

方式2：让TaskGroup函数返回组内任务或自身 #

在tg1函数中返回组内的任务（通常是组内最后一个任务），这样调用tg1()会返回可用于依赖配置的有效对象：

from datetime import datetime

from airflow.decorators import dag, task_group
from airflow.operators.dummy import DummyOperator


@dag(
    start_date=datetime(2021, 4, 20, 15, 0),
    concurrency=5,
    max_active_runs=1,
    catchup=False,
    schedule_interval="0 17 * * *"
)
def sv_load():
    start = DummyOperator(task_id="start")
    end = DummyOperator(task_id="end")

    @task_group(group_id="tg_test")
    def tg1():
        t1 = DummyOperator(task_id="task1")
        t2 = DummyOperator(task_id="task2")
        t1 >> t2
        # 返回组内最后一个任务，用于外部依赖串联
        return t2

    start >> tg1() >> end

dag = sv_load()

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