Airflow 3 中如何创建嵌套 TaskGroup 并执行组内任务？

结论

这种流程可以用 TaskGroup 实现，无需额外调用“执行任务组”的方法。在 DAG 构建时调用被装饰的函数，创建组内任务，再设置客户端任务组到全局任务组的依赖即可。

TaskGroup 用于组织任务层级，本身没有独立的执行过程。实际被调度的是组内的 env、price_task 和 global_task。要判断“上层任务运行了”意味着什么，需要展开任务组，查看具体的 Task Instance 状态。Airflow TaskGroups 文档

原始片段缺少有效缩进、任务状态和日志，无法据此确定唯一根因。不过，代码中有几处需要先检查。

原始代码需要检查哪些地方？

1. client_groups 没有展示初始化。如果使用列表收集任务组，应先创建 client_groups = []，否则执行到 append 时会报错。
2. global_group() 中的 client 没有明确来源。如果循环变量只存在于 all_clients_groups() 内，它在 global_group() 中不可见。若这里代表全局处理，应使用已定义的 G_CLIENT。
3. 定义装饰函数后仍需调用。仅定义 @task_group 函数不会创建其中的任务。原代码中的 define_client_group() 调用承担了这个作用，不能删除调用后不在其他地方补上。
4. return 的位置取决于缩进。如果 return client_groups 放在 for 循环内部，函数会在第一轮返回，后续客户端不会生成任务组。
5. 示例没有导入 logging 和 datetime。如果完整文件也缺少这些导入，DAG 解析会失败。

如果实际 DAG 已正常加载，这些问题可能只是摘录时遗漏的内容，不能直接认定为本次运行异常的原因。

如何组织客户端任务组和全局任务组？

可以把客户端公共流程定义一次，在外层 all_clients TaskGroup 中为每个客户端创建实例，再让全局组依赖整个外层组。

示例沿用原来的 my_common 调用方式，前提是该模块能在任务执行环境中导入，也没有假设这些工具函数支持新的参数。

import logging
from datetime import datetime, timezone

from airflow.sdk import dag, get_current_context, task, task_group
from my.utility import my_common

logger = logging.getLogger(__name__)
G_CLIENT = "Global_Client"


@dag(
    dag_id="my_nested_task_groups",
    schedule=None,
    start_date=datetime(2024, 4, 22, tzinfo=timezone.utc),
    catchup=False,
    is_paused_upon_creation=False,
)
def run_task_groups():
    @task_group()
    def define_client_group(client: str):
        @task(params={"client": client})
        def env():
            my_common.setup_env()

        @task(params={"client": client})
        def price_task():
            runtime_client = get_current_context()["params"]["client"]
            logger.info("Running price_task for client=%s", runtime_client)
            my_common.execute_price_task()

        env() >> price_task()

    @task_group(group_id="all_clients")
    def all_clients_groups():
        client_list = ["client_1", "client_2"]
        for client in client_list:
            define_client_group.override(
                group_id=f"{client}_group"
            )(client=client)

    @task_group(group_id="global_group")
    def global_group():
        @task(params={"client": G_CLIENT})
        def env():
            my_common.setup_env()

        @task(params={"client": G_CLIENT})
        def global_task():
            runtime_client = get_current_context()["params"]["client"]
            logger.info("Running global_task for client=%s", runtime_client)
            my_common.execute_global_task()

        env() >> global_task()

    all_clients_groups() >> global_group()


my_nested_task_groups = run_task_groups()

示例中：

• all_clients_groups() 使用 @task_group 创建了一个包含客户端子组的外层 TaskGroup。
• define_client_group.override(...)(client=client) 为每个客户端创建独立实例，并设置不同的 group_id。
• env() >> price_task() 创建两个任务并建立依赖。
• 两个任务组函数都没有显式返回值，调用结果可用于建立组间依赖，无须维护 client_groups 列表。

默认情况下，任务 ID 带有所属组的前缀，例如 all_clients.client_1_group.price_task。上述依赖会让全局组的入口任务等待各客户端组的末端任务；在默认 all_success 规则下，所有客户端的 price_task 成功后，才会执行全局组的 env。TaskGroup 层级与依赖说明

原来的 all_clients_groups() 也可以保留为普通 Python 函数，在其中创建任务组并返回列表；普通函数本身不会形成外层 TaskGroup。返回列表供调用方建立依赖，组内任务不靠返回列表启动。

params 是否会自动传给业务函数？

不会。@task(params={"client": client}) 设置的是任务运行时配置，不会自动把 client 注入 my_common.execute_price_task()，也不会自动为工具模块切换客户端。

可以在任务执行时读取：

runtime_client = get_current_context()["params"]["client"]

Airflow 3 提供了 airflow.sdk.get_current_context() 来访问任务上下文。Airflow 3 公共接口

完整示例只把读取结果用于日志。业务函数仍需自行读取上下文，或者由任务包装函数按照其真实接口显式传参。由于没有提供 my_common 的实现，无法判断现有函数是否已经处理了客户端配置。

任务级 params 还可能被手动触发时传入的配置覆盖，具体取决于部署配置。如果客户端身份必须固定，适合把它作为明确的 TaskFlow 函数参数传入，并据此调用业务逻辑。Params 文档

env() 设置的环境能否被后续任务继承？

如果 setup_env() 只是修改当前进程的 os.environ、模块全局变量或内存对象，后续任务不能依赖这些修改。Airflow 任务相互隔离，可能运行在不同进程或机器上；env() >> price_task() 只保证执行顺序。

应根据 setup_env() 的实际行为处理：

• 如果只是初始化进程环境，在执行业务逻辑的同一个任务中调用它。
• 如果创建的是数据库记录或其他外部资源，可以保留为独立任务，但后续任务必须能访问这些资源。
• 如果要传递少量配置或结果，让上游任务返回可序列化值，再交给下游任务；大量数据保存到外部存储，只传递位置或标识。

全局任务组同样不会因为依赖关系而自动收到客户端组的数据。需要显式设计数据传递，或让它读取前面任务写入的共享存储。XCom 与任务隔离说明

修改结构后仍然不执行，怎么定位？

先手动触发一个新的 DAG Run。schedule=None 不会自动创建定时运行，is_paused_upon_creation=False 也不等于立即触发运行。

在该次运行的 Graph 中展开客户端组，查看具体任务状态：

这些状态代表不同的调度或执行阶段，不能仅凭任务组外观判断内部代码是否运行。Task Instance 状态说明

如果具体任务已进入 running，应查看该任务自己的日志。DAG 文件顶层的 print(f'__name__ is {__name__}') 发生在模块加载时，不能证明 price_task() 或 global_task() 的业务代码已经执行。

备注：内容仅供参考。