Airflow run tasks at different times in the same dag?

Question

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.

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.

I can break them up into individual dags but that seems silly and 30 tasks will grow to a bigger number in the future.

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.

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.

Edit:

After y2k-shubham's explanation, I tried to implement it. But it's still not working. Fast task starts at 2019-01-30 00, finishes in a sec, and does not start 2019-01-30 01 because the slow task is still running. If possible, it'd be ideal to run 2019-01-30 01, 2019-01-30 02, 2019-01-30 03...in parallel if possible

Adding code example

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 >> slow_task # not working

Answer 1

Turns out there are two variables that can be set that will solve my problem very easily.

concurrency and max_active_runs

In the below example, you can have 4 dags running and each dag can have 4 tasks running at the same time. Other combinations are also possible.

dag = DAG(
    dag_id='sample_dag',
    default_args=default_args,
    schedule_interval='@daily',
    # this will allow up to 16 tasks to be run at the same time
    concurrency=16,
    # this will allow up to 4 dags to be run at the same time
    max_active_runs=4,
)

Answer 2

I can think of 3 possible solutions to your woes (will add more alternatives when they come to mind)

1. Set start_date on individual tasks within the DAG (apart from a start_date of DAG itself) as told here. However I would never favour this approach because it would be like a step back onto the same time-based crons that Airflow tries to replace.

2. Use pools to segregate tasks by runtime / priority. Here's an idea (you might need to rework as per your requirements): Put all tiny tasks in tiny_task_pool and all big ones in big_task_pool. Let the tiny_task_pool have significantly higher number of slots than big_task_pool. That would make starvation of your tiny-tasks much less likely. You can go creative with even more levels of pools.

3. Even if your tasks have no real dependencies between them, it shouldn't hurt much to deliberately introduce some dependencies so that all (or most) big tasks are made downstream of tiny ones (and hence change structure of your DAG). That would dub into a shortest-job-first kind of approach. You can also explore priority_weight / priority_rule to gain even more fine-grained control.

All the above alternatives assume that tasks' lengths (duration of execution) are known ahead of time. In real-world, that might not be true; or even if it is, it might gradually change overtime. For that, I'd suggest you to tweak your dag-definition script to factor-in the average (or median) runtime of your tasks over last 'n' runs to decide their priority.

• For start_date method, just supply a later start_date (actually same date, later time) to tasks that ran longer in previous runs
• For pools method, move tasks around different pools based on their previous running durations
• For task-dependency method, make longer running tasks downstream. This might sound difficult but you can visualize it like this: Create 3 DummyOperators and link them up (one after another). Now you have to fill-in all small tasks between the first 2 DummyOperators and the big ones between the next two.

Answer 3

This is likely because you have fewer execution slots than you have slow jobs. The scheduler doesn't particularly care what order it's running the tasks in, because you've said you don't care either.

If it really matters to you, these should probably be broken up into different dags or you should declare dependencies that you want the cheaper tasks to finish first. There are any number of ways to express what you want, you just have to figure out what that is.