不使用装饰器实现Airflow Task Group映射的正确方法（解决partial报错）

你遇到的AttributeError: 'function' object has no attribute 'partial'，原因是你定义的DeliveryGroup是普通Python函数，并非Airflow官方支持的可映射Task Group对象（比如用@task_group装饰的Task Group或TaskGroup子类），因此不支持.partial()和.expand()方法。

方式一：手动遍历生成多个Task Group实例（最直接） #

直接遍历目标字典列表，为每个目标创建独立的Task Group实例，确保每个Group的group_id唯一即可：

# 获取目标配置列表
destinations = dag.default_args.get('destinations')

# 遍历每个目标，生成对应Task Group
for idx, deliver_args in enumerate(destinations):
    # 为每个Group设置唯一ID，避免冲突
    deliver_group = DeliveryGroup(
        group_id=f'deliver_group_{idx}',
        file="my_file.csv",
        deliver_args=deliver_args
    )
    # 按需设置任务依赖，例如：
    # previous_task >> deliver_group

方式二：基于TaskGroup.partial()实现动态映射 #

若想使用Airflow的动态映射语法（.expand()），需将Task Group的任务逻辑封装为可调用函数，再结合TaskGroup.partial()构建可映射的Task Group：

1. 重构任务构建逻辑：

from airflow.utils.task_group import TaskGroup
from airflow.operators.python import BranchPythonOperator
from airflow.operators.email import EmailOperator
from airflow.providers.sftp.operators.sftp import SFTPOperator
from airflow.providers.sftp.hooks.sftp import SFTPHook

def build_delivery_tasks(tg: TaskGroup, file: str, deliver_args: dict):
    # 目标类型选择任务
    selector_task = BranchPythonOperator(
        task_id='destination_selector',
        python_callable=lambda: f"{deliver_args.get('type')}",
        task_group=tg
    )
    
    # 邮件发送任务
    email_task = EmailOperator(
        task_id="email",
        to=deliver_args.get('to'),
        subject=deliver_args.get('subject'),
        cc=deliver_args.get('cc'),
        html_content=deliver_args.get('body'),
        files=[file],
        task_group=tg
    )

    # SFTP传输任务
    sftp_task = SFTPOperator(
        task_id="sftp",
        sftp_hook=SFTPHook(
            remote_host=deliver_args.get('host'),
            username=deliver_args.get('username'),
            password=deliver_args.get('password'),
            port=deliver_args.get('port', 22)),
        local_filepath=[file],
        remote_filepath=[deliver_args.get('path')],
        task_group=tg
    )

    selector_task >> [email_task, sftp_task]

2. 使用动态映射创建Task Group：

deliver_groups = TaskGroup.partial(
    group_id='deliver',
    file="my_file.csv",
    build_task_group_fn=build_delivery_tasks
).expand(
    deliver_args=dag.default_args.get('destinations')
)

• 每个Task Group的group_id必须唯一，否则Airflow会抛出重复ID错误；
• 若使用动态映射，Airflow版本需在2.3及以上（该版本开始支持Task Group的动态映射）；
• 手动遍历的方式更直观，适合大多数场景，而动态映射语法更简洁，适合复杂的参数化场景。

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