GitHub user raphaelauv edited a comment on the discussion: MappedTaskGroup
without taskflow
"solution" until a MappedTaskGroup is available in the task-sdk
```python
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
from typing import Callable
import time
def mapped_taskgroup_builder(tg_callable: Callable, group_id: str, **tg_kwargs):
from airflow.sdk import task_group
tg = task_group(group_id=group_id, **tg_kwargs)
return tg(tg_callable)
with DAG(
dag_id="group_dag",
schedule=None,
):
first = PythonOperator(
task_id="first",
python_callable=lambda: [1, 10, 20, 30]
)
def sleep(duration):
time.sleep(duration)
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
mapped_taskgroup_builder(group_id="g1", tg_callable=tg1,
tooltip="SUPER").expand(duration=first.output)
```
GitHub link:
https://github.com/apache/airflow/discussions/50297#discussioncomment-17510964
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
