title: Airflow run tasks at different times in the same dag?
body: <p>I have 30 individual tasks in a dag, they have no dependencies between each other. The tasks run the same code. The only difference is the data volume, some tasks will finish in secs, some tasks will take 2 hours or more.</p>
  
  <p>The problem is during catchup, the tasks that finish in secs are blocked by tasks that take hours to finish before they move on to the next execution date.</p>
  
  <p>I can break them up into individual dags but that seems silly and 30 tasks will grow to a bigger number in the future.</p>
  
  <p>Is there any way to run tasks in the same dag at different execution times? Like as soon as a task finish, take on the next execution date, regardless of how other tasks are doing.</p>
  
  <p>Adding pic for illustration. Basically, I'd like to see two more solid green boxes on the first row while the third row is still running behind.</p>
  
  <p><a href="https://i.stack.imgur.com/IqGo2.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/IqGo2.png" alt="airflow_dag_ideal"></a></p>
  
  <p>Edit:</p>
  
  <p>After y2k-shubham's <a href="https://stackoverflow.com/a/54587355/5004780">explanation</a>, I tried to implement it. But it's still not working. Fast task starts at <code>2019-01-30 00</code>, finishes in a sec, and does not start <code>2019-01-30 01</code> because the slow task is still running. If possible, it'd be ideal to run <code>2019-01-30 01</code>, <code>2019-01-30 02</code>, <code>2019-01-30 03</code>...in parallel if possible</p>
  
  <p>Adding code example</p>
  
  <pre><code>import time
  from datetime import datetime
  
  from airflow import DAG
  from airflow.operators.python_operator import PythonOperator
  from airflow.utils.trigger_rule import TriggerRule
  
  default_args = {
      'owner': 'test',
      'depends_on_past': False,
      'start_date': datetime(2019, 1, 30, 0, 0, 0),
      'trigger_rule': TriggerRule.DUMMY
  }
  
  dag = DAG(dag_id='test_dag', default_args=default_args, schedule_interval='@hourly')
  
  
  def fast(**kwargs):
      return 1
  
  
  def slow(**kwargs):
      time.sleep(600)
      return 1
  
  
  fast_task = PythonOperator(
      task_id='fast',
      python_callable=fast,
      provide_context=True,
      priority_weight=10000,
      pool='fast_pool',
      # weight_rule='upstream', # using 1.9, this param doesn't exist
      dag=dag
  )
  
  slow_task = PythonOperator(
      task_id='slow',
      python_callable=slow,
      provide_context=True,
      priority_weight=500,
      pool='slow_pool',
      # weight_rule='upstream', # using 1.9, this param doesn't exist
      dag=dag
  )
  
  fast_task &gt;&gt; slow_task # not working
  </code></pre>