Airflow Tasks分组[编辑 | 编辑源代码]

介绍[编辑 | 编辑源代码]

在Apache Airflow中，Tasks分组（Task Groups）是一种将多个相关任务组织成逻辑单元的功能，类似于编程中的“函数”或“模块”。通过分组，可以简化DAG（有向无环图）的结构，提高可读性和维护性，尤其适用于复杂的工作流。

分组后的任务在Airflow UI中会显示为可折叠的节点，支持嵌套结构。这对于以下场景特别有用：

将重复的子流程封装为可复用的单元
减少DAG视觉复杂度
实现层次化的任务管理

基础语法[编辑 | 编辑源代码]

使用 TaskGroup 上下文管理器创建分组（Airflow 2.0+）：

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils.task_group import TaskGroup

with DAG('example_task_group', schedule_interval=None) as dag:
    start = DummyOperator(task_id='start')

    with TaskGroup(group_id='process_data') as process_group:
        task_a = DummyOperator(task_id='task_a')
        task_b = DummyOperator(task_id='task_b')
        task_a >> task_b

    end = DummyOperator(task_id='end')

    start >> process_group >> end

输出效果[编辑 | 编辑源代码]

在Airflow UI中，process_data组会显示为一个可展开的节点，包含内部的task_a和task_b。

嵌套分组[编辑 | 编辑源代码]

Task Groups支持多级嵌套，适合模块化设计：

with DAG('nested_groups', schedule_interval=None) as dag:
    with TaskGroup(group_id='outer_group') as outer:
        with TaskGroup(group_id='inner_group') as inner:
            task1 = DummyOperator(task_id='task1')
            task2 = DummyOperator(task_id='task2')
            task1 >> task2

实际案例：ETL流水线[编辑 | 编辑源代码]

假设有一个ETL（Extract-Transform-Load）流程，其中转换阶段包含多个并行任务：

对应代码实现：

with DAG('etl_pipeline', schedule_interval='@daily') as dag:
    extract = DummyOperator(task_id='extract')

    with TaskGroup(group_id='transform') as transform:
        clean = DummyOperator(task_id='clean_data')
        validate = DummyOperator(task_id='validate_data')
        aggregate = DummyOperator(task_id='aggregate_data')

    load = DummyOperator(task_id='load')

    extract >> transform >> load

高级特性[编辑 | 编辑源代码]

依赖传递[编辑 | 编辑源代码]

分组内的任务可以跨组建立依赖：

with TaskGroup(group_id='group1') as g1:
    t1 = DummyOperator(task_id='t1')

with TaskGroup(group_id='group2') as g2:
    t2 = DummyOperator(task_id='t2')

t1 >> g2  # 组间依赖

动态分组[编辑 | 编辑源代码]

通过循环动态生成分组（需Airflow 2.3+）：

for region in ['us', 'eu', 'asia']:
    with TaskGroup(group_id=f'process_{region}') as tg:
        fetch = DummyOperator(task_id=f'fetch_{region}')
        parse = DummyOperator(task_id=f'parse_{region}')
        fetch >> parse