动态任务组vs动态有向无环图

在Airflow中，可以使用动态任务组或动态有向无环图来构建动态工作流，以处理在运行时生成的任务。动态任务组是一组在父任务中手动定义的子任务，而动态有向无环图(DAG)是使用代码动态生成的任务图。

以下是Airflow中使用动态任务组和动态DAG的示例：

动态任务组示例：

from airflow import DAG from airflow.operators.bash_operator import BashOperator from datetime import datetime

default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2019, 1, 1), 'email_on_failure': False, 'email_on_retry': False, 'retries': 1 }

dag = DAG('dynamic_task_group_example', default_args=default_args, schedule_interval='@once')

def create_dynamic_task_group(task_list): task_group = [] for task in task_list: task_group.append(BashOperator(task_id=task, bash_command='echo {}'.format(task), dag=dag)) return task_group

task_list = ['task1', 'task2', 'task3']

dynamic_task_group = create_dynamic_task_group(task_list)

dynamic_task_group

这个例子中，我们使用一个函数来动态地创建一个由BashOperator任务构成的任务组，这些任务的ID和Bash命令根据给定的任务列表动态生成。我们可以将这个任务组添加到DAG中，然后运行这个DAG。

动态DAG示例：

from airflow import DAG from airflow.operators.bash_operator import BashOperator from datetime import datetime

def create_dynamic_dag(dag_id, schedule, default_args):

dag = DAG(dag_id,
          default_args=default_args,
          schedule_interval=schedule)

with dag:
    task1 = BashOperator(
                task_id='task1',
                bash_command='echo "Hello from task1"',
                dag=dag)
        
    task2 = BashOperator(
                task_id='task2',
                bash_command='echo "Hello from task2"',
                dag=dag)

    # Generate tasks dynamically
    for i in range(3):
        task = BashOperator(
                    task_id='dynamic_task_{}'.format(i),
                    bash_command='echo "Hello from dynamic task {}"'.format(i),
                    dag=dag)
        task >> task2

return dag

default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2019, 1, 1), 'email_on_failure': False, 'email_on_retry': False, 'retries': 1 }

dag_id = 'dynamic_dag_example' schedule = '@once'

dynamic_dag = create_dynamic_dag(dag_id, schedule, default_args)

通过这个例子，我们可以看到如何使用一个函数来动态地生成Airflow任务，然后在DAG中将