Airflow exécute des tâches à des moments différents dans le même jour?

J'ai 30 tâches individuelles dans un dag, elles n'ont aucune dépendance entre elles. Les tâches exécutent le même code. La seule différence est le volume de données, certaines tâches se termineront en quelques secondes, certaines tâches prendront 2 heures ou plus.

Le problème est pendant le rattrapage, les tâches qui se terminent en secondes sont bloquées par des tâches qui prennent des heures avant d'être terminées. passer à la prochaine date d'exécution.

Je peux les diviser en points individuels, mais cela semble idiot et 30 tâches deviendront un plus grand nombre à l'avenir.

Y en a-t-il moyen d'exécuter des tâches dans le même temps à des moments d'exécution différents? Comme dès qu'une tâche est terminée, prenez la prochaine date d'exécution, quelle que soit la façon dont les autres tâches se déroulent.

Ajout d'une image à titre d'illustration. En gros, j'aimerais voir deux autres boîtes vertes pleines sur la première ligne alors que la troisième ligne est toujours en retard.

Modifier :

Après l ' explication de y2k-shubham, j'ai essayé de l'implémenter. Mais ça ne marche toujours pas. La tâche rapide commence le 2019-01-30 00 , se termine dans une seconde et ne démarre pas le 2019-01-30 01 car la tâche lente est toujours en cours d'exécution. Si possible, il serait idéal d'exécuter 2019-01-30 01 , 2019-01-30 02 , 2019-01-30 03 code > ... en parallèle si possible

Ajout d'un exemple de 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 >> slow_task # not working