apache / apache/airflow

Dynamically Change Tasks Queues Using DAG Parameters

Open
#35,542 5 comments 0 reactions 0 assignees View on GitHub
area:scheduler kind:feature
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.