Dynamic queueing via REST API trigger

[FelixReuthlingerBMW]

Description

I want to run Airflow DAGs triggered using the REST API in different queue and want to set the queue name on calling the REST API.

Use case / motivation

Currently Airflow does support to set queues only on a per DAG deployment level, but in a case where I have a DAG that I want once run with high-prio and 1000s of times with a lower prio, I don't have a chance to make this happen using the REST API (which we need to use for doing programmatic integration with other software / services.

A work around is to deploy the DAG twice with some different hard-coded value for setting the queue, but this will cause huge amount of code and/or code deployment duplications. Also probably causing naming conflicts, etc. Applying work arounds for the part that a feature is missing that I can set the queue for a DAG run using the REST API's data model.

I would want to just send a Json like this to Airflow and it will figure out how to queue the DAG run:

{
"conf": {
... my application config for the DAG run...
}
"queue": "..."
}

Related Issues

Not aware of, yet.

[mik-laj]

This is not officially supported, but you can try to use a cluster policy to configure the queue based on DAGRun.conf. See:
https://airflow.readthedocs.io/en/latest/concepts.html#mutate-task-instances-before-task-execution

[FelixReuthlingerBMW]

Thanks for the hint. If I understand it correctly, this would just apply different settings to single tasks, but not the whole DAG run, right?

Problem then would be, if I first start 1000 DAG runs, the high-prio DAG run queuing as number 1001 would still need to wait until all the other DAG runs were finished, independently from in which queue the tasks are executed, right?

[mik-laj]

@FelixReuthlingerBMW Yes. But you can assign all tasks from a given DAG to one queue. In Airflow, tasks are assigned to a queue, not a DAG Run.

No. The tasks will run according to the queues to which they have been assigned. You can create two queues to run some tasks faster.

[FelixReuthlingerBMW]

Maybe I should state that we are using limitations for parallel runs:

dag:
concurrency: 60
max_active_runs: 40

So, this limits to having maximal 40 DAG runs in parallel. I guess this then would not start more DAG runs, even if some of their tasks would be set to different queues, right?

[mik-laj]

I am not sure if active_dag_runs applies to externally trigger DAG Runs. It would have to be checked.

[FelixReuthlingerBMW]

It does, works pretty fine ;)

If triggered via REST API, a DAG run will be created, but since the max DAG runs is limited, it will not start queuing tasks, since the amount of parallel runs is already limited.