Celery & Docker: how to add custom routing?

[VladX09]

Hi!
Is it possible to add custom celery task routing into celery-docker-executor?

We can send config_source parameter via executor config but it seems like celery tasks for all pipelines and solids have the same name (execute_step_docker).

If I'm understanding correctly Celery uses this name to route tasks and though we can send task_routes dict to Celery config from Dagster, it won't actually work.

Or there is another way to configure tasks routing?
The main reason for doing this is to separate long and heavy tasks into separate workers and guarantee that other workers will be free from such tasks

[snoopfab]

Hello.

You can add a tag to your ops to specify the celery queue that will be used to dispatch the ops on celery workers :

@op(tags={'dagster-celery/queue': 'long_working_ops'})
def long_working_op(context) :    
   ...
@op(tags={'dagster-celery/queue': 'quick_working_ops'})
def quick_working_op(context) :    
   ...

and start your celery workers with the -q flag
for example with docker compose :

entrypoint: "dagster-celery worker start --loglevel=info --app=dagster_celery.app -y celeryconfig.yaml  -q long_working_ops --name long-working-worker"

Hope it helps,

[VladX09]

Thank you very much!