Dynamically Change Tasks Queues Using DAG Parameters
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### Description
It is not possible at the moment to dynamically re-route tasks depending on the DAG Parameters.
An example:
```
@task
def helloWorld():
print("Hello World!")
@dag(
default_args=DEFAULT_ARGS,
dag_id='python_task',
description='Test Python Task',
start_date=datetime.now(),
catchup=False,
params={
"helloWorldQueue": "my-queue"
}
)
def rsgPythonTask():
helloWorld.override(queue="{{ params.helloWorldQueue }}")()
```
This is quite an important feature in a Task Scheduler and I currently cannot find another way of doing it.
I've also tried to use cluster_policies, but the task_policy does not allow to change the queue, and the task_mutation_policy somehow hangs my entire Airflow setup and does not work correctly.(2.5.3).
### Use case/motivation
We do have a Core Airflow instance with Localized workers (India, UK, etc...)
We run Airflow on top of a Job Automation system that creates jobs based on custom configurations and finally triggers a DAG linked to this Configuration.
Some Tasks **must** run in a localized way, and we want to be able to re-route tasks to the localized worker automatically using the Configuration. The Configuration can feed the location of the Worker by using the "queue" parameter, so each task can be re-reouted to a specific queue in a specific location.
I am not entirely sure why it's not possible to do this now. It's not sustainable and not scalable at all to have to hardcode queue names for workers in the DAG code. Unless I am missing something, and this is currently doable somehow.
### Related issues
_No response_
### Are you willing to submit a PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
The issue names no files or tests; start by tracing how @task queue overrides, DAG parameters, and the cluster policy paths handle queue values. Compare the task_policy and task_mutation_policy behavior described in the report. Done should allow a DAG parameter to route tasks to localized queues without hardcoding queue names.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, data-engineering
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100