MappedTaskGroup without taskflow

[raphaelauv]

Hi I'm trying to use dynamic task mapping on a taskgroup without the taskflow syntax

this is working ->

from datetime import timedelta
import time
from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.decorators import task_group
from pendulum import today

with DAG(
        dag_id="dag_group_DTM",
        start_date=today("UTC").add(days=-1),
        schedule=timedelta(days=1),
):

    first = PythonOperator(
        task_id="first",
        python_callable=lambda : [1, 10, 20, 30]
    )


    def sleep(duration):
        time.sleep(duration)

    @task_group(group_id="group_1")
    def tg1(duration):
        a = PythonOperator(task_id="a", python_callable=sleep, op_kwargs={"duration": duration})
        b = PythonOperator(task_id="b", python_callable=sleep, op_kwargs={"duration": duration})
        c = PythonOperator(task_id="c", python_callable=sleep, op_kwargs={"duration": duration})
        a >> b >> c

    tg1.expand(duration=first.output)

but I would like to not mix traditional syntax with taskflow syntax, and I did not yet found a way using

from airflow.utils.task_group import MappedTaskGroup

do you have any idea if it's possible ? thanks all

[pykenny]

From documentation it's not recommended to use MappedTaskGroup directly (reference) so I don't think it's not a good idea.

If you want to avoid decorator syntax in DAG definition, it may be better to wire task inputs/outputs just at task level instead of using a mapped task group. What about giving this adjustment a try? Looks a bit redundant but I think it is logically equivalent to the original one.

...
from airflow.utils.task_group import TaskGroup

...

with DAG(
        dag_id="dag_group_DTM",
        start_date=today("UTC").add(days=-1),
        schedule=timedelta(days=1),
):

    first = PythonOperator(
        task_id="first",
        python_callable=lambda : [1, 10, 20, 30]
    )

    def sleep(duration):
        # It's recommended to place module imports in the task function (subroutine),
        # so that all required imports only happens when the task gets run.
        import time

        time.sleep(duration)

    with TaskGroup(group_id="group_1"):
        a = PythonOperator.partial(task_id="a", python_callable=sleep).expand(duration=duration.output)
        b = PythonOperator.partial(task_id="b", python_callable=sleep).expand(duration=duration.output)
        c = PythonOperator.partial(task_id="a", python_callable=sleep).expand(duration=duration.output)

        a >> b >> c

[raphaelauv]

This is not equivalent, in your example all A takes must end before start all B tasks

In my example when the A(1) task end the B(1) task start directly whiteout waiting

[pykenny]

I see. As for scheduling strategy mapped task group (depth-first) for sure takes approach different from mapped tasks (breadth-first).

I personally would want to have depth-first strategy by default, especially when there’s significant difference in workload between the mapped tasks but seems that they haven’t make it a general available option for now, while the decorated task group is a makeshift solution (or to say, exception) for now.