Airflow 2.5.*何时在Cloud Composer镜像中可用？需实现动态Task Group

Cloud Composer对Airflow 2.5.*的支持时间 #

Cloud Composer对Airflow新版本的适配通常会在上游版本发布后的数周到数月内完成，具体Airflow 2.5.*系列的上线时间需以Google Cloud官方发布的更新公告为准。你可以通过Cloud Composer的版本更新日志或Google Cloud控制台的版本选项来关注最新支持情况。

动态Task Group实现示例 #

在Airflow 2.5及以上版本中，可通过TaskGroup的expand()方法基于上游任务输出的列表创建动态Task Group，示例代码如下：

from airflow.decorators import dag, task, task_group
from airflow.operators.bash import BashOperator
from typing import List

default_dag_args = {
    # 补充你的默认配置参数，如owner、start_date等
}

@dag(
    'my_dag',
    schedule=None,
    catchup=False,
    default_args=default_dag_args
)
def taskflow():
    @task
    def generate_data() -> List[str]:
        return ['data1', 'data2', 'data3']
    
    @task_group
    def task_group_tasks(d):
        task_1 = BashOperator(task_id=f'task_1_{d}', bash_command='exit 0')
        task_2 = BashOperator(task_id=f'task_2_{d}', bash_command='exit 0')
        task_1 >> task_2

    list_data = generate_data()
    task_group_tasks.expand(data=list_data)

dag = taskflow()

这段代码中，generate_data任务生成待遍历的数据集，task_group_tasks作为Task Group模板，通过expand(data=list_data)为列表内每个元素创建独立的Task Group实例，每个实例包含有依赖关系的task_1和task_2任务。

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