编辑“︁Airflow Tasks分组”︁

= Airflow Tasks分组 =

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

分组后的任务在Airflow UI中会显示为可折叠的节点，支持嵌套结构。这对于以下场景特别有用：  
* 将重复的子流程封装为可复用的单元  
* 减少DAG视觉复杂度  
* 实现层次化的任务管理  

== 基础语法 ==  
使用 <code>TaskGroup</code> 上下文管理器创建分组（Airflow 2.0+）：

<syntaxhighlight lang="python">
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
</syntaxhighlight>

=== 输出效果 ===  
在Airflow UI中，<code>process_data</code>组会显示为一个可展开的节点，包含内部的<code>task_a</code>和<code>task_b</code>。

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

<syntaxhighlight lang="python">
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
</syntaxhighlight>

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

<mermaid>
graph LR
    A[extract] --> B{{transform}}
    B --> C[load]
    subgraph transform
        B --> D[clean_data]
        B --> E[validate_data]
        B --> F[aggregate_data]
    end
</mermaid>

对应代码实现：  
<syntaxhighlight lang="python">
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
</syntaxhighlight>

== 高级特性 ==  
=== 依赖传递 ===  
分组内的任务可以跨组建立依赖：  
<syntaxhighlight lang="python">
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  # 组间依赖
</syntaxhighlight>

=== 动态分组 ===  
通过循环动态生成分组（需Airflow 2.3+）：  
<syntaxhighlight lang="python">
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
</syntaxhighlight>

== 注意事项 ==  
1. 分组ID（<code>group_id</code>）必须全局唯一  
2. UI中的默认折叠行为可通过<code>default_args={'tooltip': '...'}</code>配置  
3. 旧版Airflow（<2.0）需使用<code>SubDagOperator</code>（已弃用）  

== 数学表达 ==  
分组可视为DAG的子树，其复杂度从<math>O(n^2)</math>降至<math>O(m \cdot k)</math>（<math>m</math>=组数，<math>k</math>=组内任务数）。  

== 总结 ==  
Task Groups是优化Airflow DAG结构的核心工具，通过：  
* 逻辑隔离降低认知负担  
* 可视化折叠提升导航效率  
* 支持嵌套和动态生成适应复杂场景  

建议在任务数超过10个或存在明显功能分区时优先使用分组。

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow DAG开发]]