SmileCaoJ opened a new issue, #23051:
URL: https://github.com/apache/airflow/issues/23051
### Apache Airflow version
2.2.4
### What happened
I want to distribute task to worker
step1. write a dag
with DAG(
dag_id='example_bash_operator_queue',
schedule_interval='0 0 * * *',
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
catchup=False,
dagrun_timeout=datetime.timedelta(minutes=60),
tags=['example', 'example2'],
params={"example_key": "example_value", "roboinstance_id":
"roboinstance_id"},
) as dag:
run_this_last = DummyOperator(
task_id='run_this_last_queue',
queue="{{ params.roboinstance_id }}",
)
# [START howto_operator_bash]
run_this = BashOperator(
task_id='run_after_loop_queue',
bash_command='echo 1',
queue="{{ params.roboinstance_id }}",
)
# [END howto_operator_bash]
run_this >> run_this_last
step2 triger a dagrun with conf
{"example_key": "example_value", "roboinstance_id": "robo123"}
but it dose not work
I see log in scheduler
[2022-04-18 15:01:17,313]
{scheduler_job.py:473} INFO - Sending
TaskInstanceKey(dag_id='example_bash_operator_queue', task_id='runme_queue_0',
run_id='manual__2022-04-18T07:01:09+00:00', try_number=1) to executor with
priority 3 and queue {{ params.roboinstance_id }}
[2022-04-18 15:01:17,313] {base_executor.py:82} INFO -
Adding to queue: ['airflow', 'tasks', 'run', 'example_bash_operator_queue',
'runme_queue_0', 'manual__2022-04-18T07:01:09+00:00', '--local', '--subdir',
'/usr/local/lib/python3.6/dist-packages/airflow/example_dags/example_bash_operator_queue.py']
[2022-04-18 15:01:17,314] {scheduler_job.py:473} INFO -
Sending TaskInstanceKey(dag_id='example_bash_operator_queue',
task_id='runme_queue_1', run_id='manual__2022-04-18T07:01:09+00:00',
try_number=1) to executor with priority 3 and queue {{ params.roboinstance_id }}
[2022-04-18 15:01:17,314] {base_executor.py:82} INFO -
Adding to queue: ['airflow', 'tasks', 'run', 'example_bash_operator_queue',
'runme_queue_1', 'manual__2022-04-18T07:01:09+00:00', '--local', '--subdir',
'/usr/local/lib/python3.6/dist-packages/airflow/example_dags/example_bash_operator_queue.py']
[2022-04-18 15:01:17,314] {scheduler_job.py:473} INFO -
Sending TaskInstanceKey(dag_id='example_bash_operator_queue',
task_id='runme_queue_2', run_id='manual__2022-04-18T07:01:09+00:00',
try_number=1) to executor with priority 3 and queue {{ params.roboinstance_id }}
### What you think should happen instead
the queue of operator will be replace of params
### How to reproduce
_No response_
### Operating System
ubuntu 18.04
### Versions of Apache Airflow Providers
_No response_
### Deployment
Other
### Deployment details
_No response_
### Anything else
_No response_
### Are you willing to submit 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
