title: How to dynamically generate and use Dummy Operators in Airflow
body: <p>I have an Airflow DAG that looks a bit like this:</p>
  <pre><code>from datetime import datetime, timedelta
  
  from airflow import DAG
  from airflow.contrib.operators.databricks_operator import \
      DatabricksRunNowOperator
  from airflow.models import Variable
  from airflow.operators.dummy_operator import DummyOperator
  
  # These args will get passed on to each operator
  default_args = {
      &quot;owner&quot;: &quot;airflow&quot;,
      &quot;depends_on_past&quot;: False,
      &quot;email_on_failure&quot;: False,
      &quot;email_on_retry&quot;: False,
      &quot;retries&quot;: 2,
      &quot;retry_delay&quot;: timedelta(seconds=30),
      &quot;is_paused_upon_creation&quot;: True,
      &quot;timeout_seconds&quot;: 604800,
  }
  
  # Primary Control Block
  with DAG(
      &quot;name&quot;,
      start_date=datetime(2021, 1, 1),
      schedule_interval=&quot;@once&quot;,
      default_args=default_args,
      catchup=False,
      max_active_runs=1,
  ) as dag:
  
      # tasks
      imp_step_1 = DatabricksRunNowOperator(...)
  
      imp_step_2 = DatabricksRunNowOperator(...)
  
      data_obj_list_1 = [&quot;a&quot;, &quot;b&quot;, &quot;c&quot;, &quot;1&quot;, &quot;2&quot;, &quot;3&quot;]
  
      data_obj_list_2 = [&quot;d&quot;, &quot;e&quot;, &quot;f&quot;]
  
      ...
      ...
      ...
  
      data_obj_list_n = [&quot;x&quot;, &quot;y&quot;, &quot;z&quot;]
      
      def generate_tasks(job):
          return DatabricksRunNowOperator(...)
  
      dummy_step_1 = DummyOperator(
          task_id=&quot;step_1&quot;,
          trigger_rule=&quot;all_success&quot;,
      )
  
      dummy_step_2 = DummyOperator(
          task_id=&quot;step_2&quot;,
          trigger_rule=&quot;all_success&quot;,
      )
  
      dummy_step_3 = DummyOperator(
          task_id=&quot;step_3&quot;,
          trigger_rule=&quot;all_success&quot;,
      )
      ...
      ...
  
      dummy_step_n = DummyOperator(
          task_id=&quot;step_n&quot;,
          trigger_rule=&quot;all_success&quot;,
      )
  
  
      imp_step_1 &gt;&gt; imp_step_2 &gt;&gt; dummy_step_1
  
      for obj in data_obj_list_1:
          dummy_step_1 &gt;&gt; generate_tasks(obj) &gt;&gt; dummy_step_2
  
      for obj in data_obj_list_2:
          dummy_step_2 &gt;&gt; generate_tasks(obj) &gt;&gt; dummy_step_3
  
      ...
      ...
  
      for obj in data_obj_list_n:
          dummy_step_n &gt;&gt; generate_tasks(obj) &gt;&gt; dummy_step_(n+1)
  </code></pre>
  <p>Basically, the idea is to have an explicit dependency on the first two steps, then dynamically generate tasks, with all the tasks in each array being together and linked by the Dummy Operator tasks.</p>
  <p>The DAG works well, except that with the exception of the first list, all the ones that follow, are broken down simply because it is more efficient to run a limited number of concurrent tasks (given the compute requirements of the items in those list as opposed to the ones in the first list). So I had the bright idea of using a single list, breaking it down into multiple lists of 3 dynamically (so even if we add more objects later on, it'd be handled automatically), and then using those sub-lists to dynamically generate the tasks, and have dynamically generated Dummy Operators linking them as well. Basically, recreate the DAG but with almost-complete dynamicism. Here's how far I got:</p>
  <pre><code>from datetime import datetime, timedelta
  
  from airflow import DAG
  from airflow.contrib.operators.databricks_operator import \
      DatabricksRunNowOperator
  from airflow.models import Variable
  from airflow.operators.dummy_operator import DummyOperator
  
  # These args will get passed on to each operator
  default_args = {
      &quot;owner&quot;: &quot;airflow&quot;,
      &quot;depends_on_past&quot;: False,
      &quot;email_on_failure&quot;: False,
      &quot;email_on_retry&quot;: False,
      &quot;retries&quot;: 2,
      &quot;retry_delay&quot;: timedelta(seconds=30),
      &quot;is_paused_upon_creation&quot;: True,
      &quot;timeout_seconds&quot;: 604800,
  }
  
  # Primary Control Block
  with DAG(
      &quot;name&quot;,
      start_date=datetime(2021, 1, 1),
      schedule_interval=&quot;@once&quot;,
      default_args=default_args,
      catchup=False,
      max_active_runs=1,
  ) as dag:
  
      # tasks
      imp_step_1 = DatabricksRunNowOperator(...)
  
      imp_step_2 = DatabricksRunNowOperator(...)
  
      data_obj_list_1 = [&quot;a&quot;, &quot;b&quot;, &quot;c&quot;, &quot;1&quot;, &quot;2&quot;, &quot;3&quot;]
  
      data_objs = [&quot;d&quot;, &quot;e&quot;, &quot;f&quot; .... , &quot;x&quot;, &quot;y&quot;, &quot;z&quot;]
  
      data_objs = [data_objs[i : i + 3] for i in range(0, len(data_objs), 3)]
  
      def generate_tasks(job):
          return DatabricksRunNowOperator(...)
  
  
      def generate_dummies(job):
          return DummyOperator(...)
  
      dummy_step_1 = DummyOperator(
          task_id=&quot;step_1&quot;,
          trigger_rule=&quot;all_success&quot;,
      )
  
      dummy_step_2 = DummyOperator(
          task_id=&quot;step_2&quot;,
          trigger_rule=&quot;all_success&quot;,
      )
  
  
      imp_step_1 &gt;&gt; imp_step_2 &gt;&gt; dummy_step_1
  
      for obj in data_obj_list_1:
          dummy_step_1 &gt;&gt; generate_tasks(obj) &gt;&gt; dummy_step_2
  
      for i in range(len(data_objs)):
          for k in data_objs[i]:
              generate_dummies(i+2) &gt;&gt; generate_tasks(k) &gt;&gt; generate_dummies(i+3)
  </code></pre>
  <p>But Airflow throws an error and says:</p>
  <pre><code>airflow.exceptions.DuplicateTaskIdFound: Task id 'step_2' has already been added to the DAG
  </code></pre>
  <p>Step_2 in this case, refers to the dummy operator step. How can I get the second version to work like the first one and avoid generating a whole bunch of Operators ahead of time, and if things change in terms of fewer or more items being added to the list, how to make it so that I wouldn't have to also go and update the number of Operators being created?</p>