Custom queue for a job/op in Dagster

[athenadataflow]

I'm trying to set up a data pipeline with multiple celery workers, RabbitMQ and Dagster.
In this picture I set up a tags for "gpu" queue but as after the job start running, all the tasks go to the default "dagster" queue.
I tried to set the tags inside op config instead of job config but no success.

I ran through the dagster-celery files and other posts but couldn't find out the reasons, all the logic are matching so here I am.
Please help me :(.

Here's the configs:

Here's the result:

Here's the job's definition code that I used:
celery_job.py

from dagster_celery import celery_executor

from dagster import job, op


@op
def not_much(
     tags={"dagster-celery/queue":"gpu"},
):
    return


@job(executor_def=celery_executor,
     tags={"dagster-celery/queue":"gpu"},
     config={
        "execution":{
            "config":{
                "backend": "db+postgresql://postgres:postgres@postgres:5433/dagster_celery",
                 "broker": "pyamqp://rabbit:rabbit@rabbit:5672//"
                }
            }
        }   
     ,
     )
def parallel_job():
    for i in range(50):
        not_much.alias("not_much_" + str(i))()`

[garethbrickman]

From the docs:

Users can configure multiple Celery queues (for example, one celery queue for each resource the user would like to limit) and multiple Celery workers per queue via the runLauncher.config.celeryK8sRunLauncher.workerQueues section of values.yaml.

Did you configure this in your values.yaml?

[athenadataflow]

I was not using Helm or Kubernetes so I haven't done this. What I did when trying to push things to specific queue in my situation was:

• Deploy the worker with "-Q" tag to a "gpu" queue. RabbitMQ recognized 1 customer on "gpu" queue.
• Set a "tags={"dagster-celery/queue":"gpu"}" in job config.

If these two steps are not sufficient and values.yaml is still necessary, how can I apply this file in this situation? Thank you for showing up, I was just about to give up <3!

[garethbrickman]

Ah I see you're working with Celery executor as in this guide not the CeleryK8sRunLauncher

[Nicomel]

Hi @athenadataflow , any follow-up or progress on this?
I'm in the same configuration, trying to setup a specific queue for some assets and it is always sending the task to the default queue whatever and wherever I set the 'dagster-celery/queue' tag