Sequential Execution of Dynamic Mapped Task

[manugarri]

Description

Dynamic tasks enable dag creation based on run time based number of inputs. This is great because it not only makes dags flexible, but also it reduces the workload on the scheduler since the tasks are defined at run time.

However, the current dynamic tasks run all inputs in parallel (via the expand function). However, there are some cases where a list of steps needs to be run sequentially, but the list is not known until runtime.

Use case/motivation

It is helpful to show an actual use case to see why the feature would be useful.

I am parsing a yaml with some EMR steps to run as a @task , and I cannot add a task inside another one with a for loop to create the actual operators that run the operations described on the yaml file.

The dag looks more or less like this :

with DAG(...):
  @task
  def parse_yaml(yaml_path):
     """ 
     read the templated yaml and expand the jinja variables inside
     This yaml has blocks of steps that have to run sequentially:
      something like this:
      groups:
        - group_name:1
          steps:
            - step_name: 1.1
              step_emr_command: spark-submit ...
            - step_name: 1.2
              step_emr_command: spark-submit ...
        - group_name:2
          steps:
             - step_name: 2.1
               step_emr_command: spark-submit ...
             - step_name: 2.2
               step_emr_command: spark-submit ...
      """
     return parsed_yaml
  
  @task_group
  def run_group(group_definition):
     setup_emr_cluster = EMRCreateJobFlowOperator()
     run_steps = EMRAddStepsOperator(group_definition['steps'])
     terminate_emr_cluster = EMRTerminateJobFlowOperator()
     setup_emr_cluster >> run_steps >> terminate_emr_cluster
  
  parsed_yaml = parse_yaml(yaml_path)
  
  # this runs all the step groups in parallel (bad)
  run_group.expand(parsed_yaml)

hopefully this pseudo example showcases what im trying to achieve. I need the run_group for group:1 to run before group2 , not in parallel.

Here is the diagram that should be run based on the yaml file above:

                          ┌───────────────────────────────────────────────────┐                ┌─────────────────────────────────────────────────┐
                          │                                                   │                │                                                 │
                          │                                                   │                │                                                 │
┌─────────────┐           │                                                   │                │                                                 │
│             │           │                                                   │                │ ┌─────────┐  ┌───────┐   ┌───────┐   ┌───────── │
│  parse_yaml ├──────────►│  ┌─────────  ┌───────┐   ┌────────┐  ┌──────────  │                │ │EMRCreate├─►│step   ├──►│step   ├──►│Terminate │
│             │           │  │EMRCreate  │ step  │   │step    │  │Terminate''e├───────────────►│ │Cluster2 │  │2.1    │   │2.2    │   │Cluster2│ │
│             │           │  │Cluster1┌─►│ 1.1   ├──►│ 1.2    ├─►│Cluster1 │  │                │ └─────────┘  └───────┘   └───────┘   └────────┘ │
└─────────────┘           │  └────────┘  └───────┘   └────────┘  └─────────┘  │                │                                                 │
                          │                                                   │                │                                                 │
                          │                                                   │                │                                                 │
                          │                                                   │                │                                                 │
                          │                                                   │                │                                                 │
                          │                                                   │                │                                                 │
                          │                             Group 1               │                │                                         Group 2 │
                          │                                                   │                │                                                 │
                          └───────────────────────────────────────────────────┘                └─────────────────────────────────────────────────┘

Related issues

There was a previous unanswered discussion on the repo (link).

There seem to be some related questions on StackOverflow ( 1, 2 3 , 4)

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[nathadfield]

Have you looked at the recent new feature that introduces the concept of Setup and Teardown tasks?

It might not give you exactly what you're after but it is certainly a better way provisioning resources as part of your DAGs.

[manugarri]

@nathadfield thanks for the info, i was not aware of that feature. I dont believe it helps regarding the dynamic sequential tasks but seems pretty interesting nonetheless :)

[potiuk]

No. Same as in the dicussion you quoted - if you want to genereate sequential list of tasks at runtime to execute, there is no such feature in Airflow - not with the Ariflow definition ot tasks. If you want N-sequentially executed, independent task, the number of them and dependencies between them (i.e. DAG structure MUST be set at parsing time) . There is no such feature in Airflow. One of the reasons is that if you want to dynamically create such task, you need to dynamically create dependencies, and that might miean that for example your dag will stop being a DAG, it will become dynamically a graph with a circle for example. that's why you canmot add dependencies dynamically, the whole DAG graph must be resolved before scheduler will start scheduling the tasks, because it has to calculate dependencies.

What you can do however (since you do not want to use parallelism feature of Airflow and distributing such sequential tasks among different nodes) - you can write your own "sequential execution task" that will use EMRC hook and will simply exectute your tasks in a loop one by one. And then then loop can be arbitrary long and dyanamic.

Im this case you will not get UI visualisation, retries. partial reruns and clearing and the like,. But you will get basic Nx task executed in sequence.

You can also emulate a bit that by assigning 1-slot pool to all tasks in a group where your tasks will be competing for pool slot. But this does not guarantee sequence of execution, all your dynamically mapped tasks will be still running technically in parallel, but with parallelism = 1 - which means one at a time but in an undefined (random) sequence.

If you also relax your "runtime" expectation to less-than runtime (ie.e. for example changing for all runs between the times when you yaml file changes - let's say once a day or once a week) then you could generate your DAG from such yaml file using Danamic DAG generaiton not Dynamic Tasks Mapping. Where you simply create tasks and set dependencies between them in the python code when your file is parsed. Roughly:

@dag
def my_dag():
   y = read_yaml().
  for task in y.tasks:
      task = EMr()
      previous_task >>  task
      previous_task = task

This is absolutely classis way og generating DAGs explained in our docs even explicitly showing yaml file: https://airflow.apache.org/docs/apache-airflow/stable/howto/dynamic-dag-generation.html#dynamic-dags-with-external-configuration-from-a-structured-data-file

But then, if it changes often and wildely, it's no good use. Such dag should change slowly - far less frequent (orders of magnitude) than the frequency of DAG runs.

And yes @nathadfield suggested well that whatever you do (if you will decide to use Airflow for this - somewhat niche in Airflow world - case for setting up and tearing down your cluster

Those are - I believe all options you have. now in Airflow.

But as usual, for those who have niche cases - if you figure out a mental model when it can be generic enough and implementable, proposals on that are welcome - my feeling is tha that caliber of a use case is somewhat calling for Airflow Improvement Proposal to prepare, because (if you want to stick to runtime properties) - it calls for a feature that will allow to have subset of DAG structure modifications that will not change properties of the graph - for example just expanding linear graph branch by injecting new tasks to such graph branch.

And BTW. Converting it into discussion. This is not an issue. This is discussion on a niche case that you have that is likely not necessary good fit for Airflow now.