title: Airflow - how to set task dependencies between iterations of a for loop?
body: <p>I am using Airflow to run a set of tasks inside for loop. The purpose of the loop is to iterate through a list of database table names and perform the following actions: </p>
  
  <pre><code>for table_name in list_of_tables:
      if table exists in database (BranchPythonOperator)
          do nothing (DummyOperator)
      else:
          create table (JdbcOperator)
      insert records into table (JdbcOperator, Trigger on One Success)
  </code></pre>
  
  <p>On the Web UI, this appears like: </p>
  
  <p><a href="https://i.stack.imgur.com/WAzFt.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/WAzFt.png" alt="tasks in for loop"></a></p>
  
  <p>Currently, Airflow executes the tasks in this image from top to bottom then left to right, like: <code>tbl_exists_fake_table_one</code> --> <code>tbl_exists_fake_table_two</code> --> <code>tbl_create_fake_table_one</code>, etc. </p>
  
  <p>However, the <code>insert</code> statement for <code>fake_table_two</code> depends on <code>fake_table_one</code> being updated, a dependency not captured  by Airflow currently. (Technically this dependency is captured by the order of the <code>list_of_table_names</code>, but I believe this will be prone to error in a more complex situation)</p>
  
  <p>I want all tasks related to <code>fake_table_one</code> to run, followed by all tasks related to <code>fake_table_two</code>. How can I accomplish this in Airflow? </p>
  
  <p>Full code below: </p>
  
  <pre><code>for tbl_name in list_of_table_names:
  
      # Check if table exists by querying information tables
      def has_table(tbl_name=tbl_name):
          p = JdbcHook('conn_id')
          sql =""" select count(*) from system.tables where name = '{}' """.format(tbl_name.upper())
          count = p.get_records(sql)[0][0] #unpack the list/tuple
  
          # If the query didn't return rows, branch to Create Table Task
          # otherwise, branch to Dummy Operator (Airflow requires that both branches have a task)
          if count == 0:
              return 'tbl_create_{}'.format(tbl_name)
          else:
              return 'dummy_{}'.format(tbl_name) 
  
      # run has_table python function
      exists = BranchPythonOperator(
          task_id='tbl_exists_{}'.format(tbl_name),
          python_callable=has_table,
          depends_on_past=False,
          dag=dag
      )
  
      # Dummy Operator
      dummy = DummyOperator(task_id='dummy_{}'.format(tbl_name),dag=dag,depends_on_past=False)
  
      # Run create table SQL script
      create = JdbcOperator(
          task_id='tbl_create_{}'.format(tbl_name),
          jdbc_conn_id='conn_id',
          sql = sql_parse(script_path, 'sql/sql_create/{}.sql'.format(tbl_name)), 
          depends_on_past=False,
          dag = dag
      )
  
      # Run insert or truncate/replace SQL script
      upsert = JdbcOperator(
          task_id='tbl_upsert_{}'.format(tbl_name),
          jdbc_conn_id='conn_id',
          sql = sql_parse(script_path, 'sql/sql_upsert/{}.sql'.format(tbl_name)),
          trigger_rule=TriggerRule.ONE_SUCCESS,
          dag = dag
      )
  
      # Set dependencies
      exists &gt;&gt; create &gt;&gt; upsert 
      exists &gt;&gt; dummy &gt;&gt; upsert
  </code></pre>