dynamic create task base on dag input

[zhiqiwangebay]

The following script creates 5 tasks, but I want to dynamically set the number of tasks based on the DAG input. is it possible?

from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime

def create_dynamic_tasks(task_group, num_tasks):
    pre = None
    for i in range(num_tasks):
        t = DummyOperator(
            task_id=f"dynamic_task_{i}",
            task_group=task_group,
        )
        if pre is not None:
            pre >> t
        pre = t

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

with DAG(
    'example_dynamic_tasks',
    default_args=default_args,
    description='A simple dynamic task creation DAG',
    schedule_interval=None,
    params={'count': 1}
) as dag:

    start = DummyOperator(
        task_id='start',
    )

    end = DummyOperator(
        task_id='end',
    )

    with TaskGroup(group_id='dynamic_tasks_group') as tg1:
        create_dynamic_tasks(tg1, 5)  # Create 5 dynamic tasks

    start >> tg1 >> end

[potiuk]

Yes. You need to use dynamic task mapping https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/dynamic-task-mapping.html. and have one single task to produce array of n elements that will be input to dynamicly mapped task (and it will create n tasks).

[zhiqiwangebay]

thanks, but It seems that the task mapping is set to execute tasks in parallel, but I want to execute them sequentially.

[potiuk]

See https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/dynamic-task-mapping.html#placing-limits-on-mapped-tasks you can set limits to run 1 mappeed task at a time. This will effectively achieve sequential processing (though you will have no control over the sequence.

[potiuk]

As a general rule - Dynamic Task mapping does not allow you to impact structure of the DAG dynamically, the "structure" (i.e. tasks and dependencies) have to be known at parse time and they should not change at execution time. The Dynamic Task Mapping is a "hack" - it does not actually change the structure of the DAG, it just multiplies one task (or one task group) that is already placed in the structure. If you want to dynamically change the structure of the DAG - i.e. add new dependencies between tasks or add new tasks (rather than clone existing task to have multiple copies) then no - it cannot be done dynamically.

[zhiqiwangebay]

Thank you very much. I spent a lot of time looking for a method to dynamically create tasks. I might need to think about some workaround.