Dagster Celery Integration

Warnings

• gotcha The `dagster-celery` library requires an externally provisioned and managed Celery cluster (broker, backend, and workers). Dagster itself does not manage the lifecycle of Celery components. Misconfiguration of `broker_url` or `result_backend` in `celery_config`, or failure to start Celery workers, are common sources of errors.
  Fix: Ensure your Celery broker and result backend URLs are correct and accessible from both Dagster and Celery workers. Verify Celery workers are running and configured to listen on the correct queues (e.g., `task_default_queue`). Use `celery inspect ping` to check worker connectivity.
• breaking Dagster's execution engine configuration has evolved across major versions. Older patterns involving `config_schema` or directly passing configuration dictionaries to `executor_def` might be deprecated or behave differently. Ensure you are using the `executor_def.configured(config)` pattern.
  Fix: Always use the `executor_def=celery_executor.configured(celery_config)` pattern when defining your jobs or `Definitions` object. Refer to the official Dagster documentation for the specific version you are using.
• gotcha Version compatibility between `dagster` core and `dagster-celery` is critical. Mismatched major versions can lead to runtime errors or unexpected behavior due to API changes in the core execution engine.
  Fix: Always install `dagster-celery` with a version that matches your `dagster` core version's major and minor numbers. For example, if `dagster` is `1.13.0`, `dagster-celery` should be `0.29.0` (as per the release notes, library versions are `core_minor_version - 100` for the major part and `core_patch_version` for the minor part after Dagster 1.0).
• gotcha When using `celery_k8s_executor`, the Kubernetes cluster must be properly configured for Celery workers, including appropriate images, resource requests/limits, environment variables, secrets (for broker/backend), and persistent storage if needed. Issues with any of these can prevent ops from running or reporting results.
  Fix: Thoroughly review the `dagster-celery` Kubernetes documentation. Ensure your Celery worker deployment correctly references the image containing your Dagster code, has network access to the broker/backend, and possesses necessary permissions and resources within Kubernetes.

Install

• pip install dagster-celery Install dagster-celery

Imports

• celery_executor

  from dagster_celery import celery_executor

  Main executor for running Dagster ops on a Celery cluster.
• celery_k8s_executor

  from dagster_celery import celery_k8s_executor

  Specialized executor for running Dagster ops on a Celery cluster within Kubernetes.

Quickstart

This quickstart demonstrates how to configure a Dagster job to use the Celery executor. It defines a simple op and a job, then configures `celery_executor` with basic broker and backend URLs (using environment variables for flexibility). For this to run, you need a running Celery broker/backend (e.g., Redis) and a Celery worker started with `celery -A your_module worker -l info` pointing to your Dagster code. The job will then submit its op execution to the Celery queue.

import os
from dagster import Definitions, job, op
from dagster_celery import celery_executor

# Configure Celery broker and backend URLs via environment variables
# Example for local development with Redis: export CELERY_BROKER_URL="redis://localhost:6379/0"
# Example for local development with Redis: export CELERY_RESULT_BACKEND="redis://localhost:6379/0"
celery_broker_url = os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0')
celery_result_backend = os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/0')

celery_config = {
    "broker_url": celery_broker_url,
    "result_backend": celery_result_backend,
    "task_default_queue": "dagster_celery_queue", # Optional: dedicated queue for Dagster tasks
}

@op
def my_celery_op(context):
    context.log.info("Executing op on Celery worker!")
    return "Hello from Celery!"

@job(executor_def=celery_executor.configured(celery_config))
def my_celery_job():
    my_celery_op()

# To run this, you would need to:
# 1. Start a Redis server (or other message broker/backend).
# 2. Start a Celery worker: celery -A dagster_celery.app worker -l info -Q dagster_celery_queue
#    (Assuming your Dagster code is in a module named 'dagster_celery.app' or similar)
# 3. Load the repository with 'dagster dev' and launch a run for 'my_celery_job'.

defs = Definitions(
    jobs=[my_celery_job]
)