Воздушный поток: расставьте приоритеты в Dags

Question

Я извлекаю категорию из таблицы метаданных и создаю динамические символы для каждой категории, используя скрипт Python. Сейчас у нас около 15 категорий, поэтому у каждой категории будет свой знак. В моем файле Dag есть 3 задачи, и он выполняется последовательно.

Использование LocalExecutor. Все 15 дагов (даг-прогонов) запускаются параллельно. У нас недостаточно ресурсов (задачи тяжелые), чтобы запустить все 15 дагов параллельно.

Есть ли способ расставить приоритеты у даг-ранов? Сначала должны бежать 5 дагов, потом следующие пять и так далее. Задания должны выполняться на основе доступных ресурсов, остальные должны быть в очереди. Это должно быть динамическим.

Есть ли лучший способ исправить это? Пожалуйста, помогите.

Образец дампа:

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta


default_args = {
    'start_date': datetime(2019, 6, 03),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG('test', catchup=False, default_args=default_args, schedule_interval='*/5 * * * *')

t1 = BashOperator(
    task_id='print_start_date',
    bash_command='date',
    dag=dag)

t2 = BashOperator(
    task_id='sleep',
    bash_command='sleep 50s',
    retries=3,
    dag=dag)

t3 = BashOperator(
    task_id='print_end_date',
    bash_command='date',
    dag=dag)

t1 >> t2 >> t3

Answer 1

Не существует эффективного способа сделать это, если вы оба работаете на LocalExecutor и все они работают одновременно.

Если вы перешли на использование CeleryExecutor и у вас есть несколько рабочихЗатем вы можете использовать концепцию Очереди воздушного потока для создания «приоритетной» очереди, которая обслуживает группы обеспечения доступности баз данных, для которых вы указываете высокий приоритет.

В других вариантах можно использовать SubDAGs .Каждый из 15 DAG может быть структурирован как Sub DAG и работать в нужном вам порядке.Вот пример того, как это может выглядеть:

start ----> Sub Dag 1 --> Sub Dag  6 --> Sub Dag 11
       |--> Sub Dag 2 --> Sub Dag  7 --> Sub Dag 12
       |--> Sub Dag 3 --> Sub Dag  8 --> Sub Dag 13
       |--> Sub Dag 4 --> Sub Dag  9 --> Sub Dag 14
       |--> Sub Dag 5 --> Sub Dag 10 --> Sub Dag 15