Airflow 3嵌套Task Groups构建问题：实际任务未执行求助

本人有1-2个月Airflow 3使用经验，现遇到DAG结构相关问题。我尝试构建一个DAG，逻辑为：先为每个客户端执行一组专属任务，待所有客户端任务组完成后，再执行处理其输出数据的全局任务组。当前DAG在Airflow界面显示的结构符合预期，但执行时仅高层任务运行，实际工作任务未执行。附上基础代码结构，恳请提供排查建议。

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

print(f'__name__ is {__name__}')
logger = logging.getLogger(__name__)

G_CLIENT = 'Global_Client'

@dag(dag_id='my_nested_task_groups', schedule=None,
        start_date=datetime(2024, 4, 22),   #whatever
        catchup=False,  # we would rarely use catchup
        is_paused_upon_creation=False,  
)
def run_task_groups():

    #looping function that defines task groups
    def all_clients_groups():
        client_list = ['client_1', 'client_2']
        client_groups = []
        for client in client_list:
            
            @task_group(group_id=client+'_group')
            def define_client_group():
                
                @task(params={"client": client})
                def env():
                    my_common.setup_env()
        
                @task(params={"client": client})
                def price_task():
                    my_common.execute_price_task()
              
                env() >> price_task()
            client_groups.append(define_client_group())
        return client_groups

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

        @task(params={"client": G_CLIENT})
        def global_task():
            my_common.execute_global_task()
            
        env() >> global_task()

    
    all_clients_groups() >> global_group()

#generate the DAG
run_task_groups()

排查建议 #

• 修复未初始化变量：all_clients_groups函数内的client_groups列表未初始化就调用append，会触发运行时错误，导致客户端任务组无法正确生成，需添加client_groups = []。
• 修正变量引用错误：global_group中的任务误用了未定义的client变量，需替换为全局常量G_CLIENT，否则会引发NameError，导致全局任务组无法正常解析。
• 解决闭包变量陷阱：循环中定义任务时，@task(params={"client": client})的client可能因闭包特性导致所有任务引用同一个最终值，建议将client作为参数传入任务函数，比如：

  @task
  def env(client):
      my_common.setup_env(client)
  # 调用时传入client
  env(client) >> price_task(client)
  

• 检查任务组依赖关联：确保all_clients_groups()返回的所有任务组实例都与global_group()建立了正确依赖，可使用chain工具显式关联：

  from airflow.utils.helpers import chain
  client_groups = all_clients_groups()
  chain(client_groups, global_group())
  

• 查看Airflow日志：前往Airflow UI的日志页面，检查DAG解析、调度阶段的报错信息（如导入错误、变量未定义），这类隐藏错误会导致任务无法被调度执行。
• 验证DAG解析状态：在Airflow UI的DAG列表中检查目标DAG是否标记为“Broken”，点击进入查看具体解析错误。

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