Can you dynamically expand multiple times in a DAG using Airflow 2.30

Viewed 24

As of Airflow 2.3, the .expand() and .partial() syntax were introduced to allow for dynamically creating task instances for a list or a map.

Is it possible to expand multiple times? Ie: Task 1 -> [1,2,3,4,5] Task 2 expands on task 1 and creates list of lists 1, [2,2], [3,3,3] etc. respectively. Task 3 expands again to print out separately so 15 tasks.

Currently it seems that the input from task 2 gets the 5 values versus expanding a second time.

    @task
    def extract():
        return ["1", "2", "3", "4", "5"]

    @task
    def transform(value: str) -> typing.List[str]:
        return [value] * int(value)

    @task
    def load(value: str):
        print(value)

    order_data = extract()
    results = transform.expand(value=order_data)
    load.expand(value=results)

Is this possible or would I have to create the tasks in another way?

Graph Diagram of Run

0 Answers
Related