Celery (dagster-celery)¶

Quickstart¶

To get a local rabbitmq broker started and available via the default pyamqp://guest@localhost:5672, in the dagster/python_modules/libraries/dagster-celery/ directory run:

docker-compose up

To run a celery worker:

celery -A dagster_celery.app worker -l info

To start multiple workers in the background, run:

celery multi start w2 -A dagster_celery.app -l info

To execute a job using the celery-backed executor, you’ll need to set the job’s executor_def to the celery_executor.

from dagster import job
from dagster_celery import celery_executor

@job(executor_def=celery_executor)
def my_job():
    pass

celery -A dagster_celery.app flower

from celery import Celery

from dagster_celery.tasks import create_task

app = Celery('dagster', broker_url='some://custom@value', ...)

execute_plan = create_task(app)

if __name__ == '__main__':
    app.worker_main()

celery -A my_module worker --loglevel=info

API¶

dagster_celery.celery_executor ExecutorDefinition[source]¶

Config Schema:

broker (Union[dagster.StringSource, None], optional):

The URL of the Celery broker. Default: ‘pyamqp://guest@{os.getenv(‘DAGSTER_CELERY_BROKER_HOST’,’localhost’)}//’.

backend (Union[dagster.StringSource, None], optional):

The URL of the Celery results backend. Default: ‘rpc://’.

Default Value: ‘rpc://’

include (List[String], optional):

List of modules every worker should import

config_source (Union[permissive dict, None], optional):

Additional settings for the Celery app.

retries (selector, optional):

Whether retries are enabled or not. By default, retries are enabled.

Default Value:

{
    "enabled": {}
}

Config Schema:

enabled (strict dict, optional):

Default Value:

{}

disabled (strict dict, optional):

Default Value:

{}

Celery-based executor.

The Celery executor exposes config settings for the underlying Celery app under the config_source key. This config corresponds to the “new lowercase settings” introduced in Celery version 4.0 and the object constructed from config will be passed to the celery.Celery constructor as its config_source argument. (See https://docs.celeryq.dev/en/stable/userguide/configuration.html for details.)

The executor also exposes the broker, backend, and include arguments to the celery.Celery constructor.

In the most common case, you may want to modify the broker and backend (e.g., to use Redis instead of RabbitMQ). We expect that config_source will be less frequently modified, but that when solid executions are especially fast or slow, or when there are different requirements around idempotence or retry, it may make sense to execute jobs with variations on these settings.

To use the celery_executor, set it as the executor_def when defining a job:

from dagster import job
from dagster_celery import celery_executor

@job(executor_def=celery_executor)
def celery_enabled_job():
    pass

Then you can configure the executor as follows:

execution:
  config:
    broker: 'pyamqp://guest@localhost//'  # Optional[str]: The URL of the Celery broker
    backend: 'rpc://' # Optional[str]: The URL of the Celery results backend
    include: ['my_module'] # Optional[List[str]]: Modules every worker should import
    config_source: # Dict[str, Any]: Any additional parameters to pass to the
        #...       # Celery workers. This dict will be passed as the `config_source`
        #...       # argument of celery.Celery().

Note that the YAML you provide here must align with the configuration with which the Celery workers on which you hope to run were started. If, for example, you point the executor at a different broker than the one your workers are listening to, the workers will never be able to pick up tasks for execution.

CLI¶

The dagster-celery CLI lets you start, monitor, and terminate workers.

dagster-celery worker start [OPTIONS] [ADDITIONAL_ARGS]...

dagster-celery worker list [OPTIONS]

dagster-celery worker terminate [OPTIONS] [NAME]