Celery

Celery is an open-source distributed task queue system written in Python, designed to manage the execution of tasks asynchronously across multiple workers.

It operates on a producer-consumer model, where tasks are defined as units of work and are placed into a queue. These tasks are then picked up and processed by worker processes, which constantly monitor the task queue for new jobs.

Celery uses a messaging broker, such as RabbitMQ or Redis, to facilitate communication between the client that initiates the task and the worker that executes it. This system allows for high availability and horizontal scaling, making it possible to distribute work efficiently across different threads or machines.

Celery’s protocol can be implemented in various programming languages, enabling language interoperability and integration with diverse systems. For instance, Node.js and PHP have their own implementations, and HTTP endpoints can be used to connect Celery tasks with external services, further extending its versatility.

To practice, let's redo our first exercise and understand what we get new with the use of Celery.

Install

Install Celery with:

$ python -m pip install celery

Create App

Let's create a simple application to consume the messages. It will simply display the message on standard output.

broker_url = "amqp://myuser:mypassword@localhost:5672//"
task_serializer = "json"
accept_content = ["json"]
timezone = "UTC"
enable_utc = True

from celery import Celery

app = Celery("my_ml_app")

# Load the Celery configuration from celeryconfig.py
app.config_from_object("celeryconfig")


@app.task
def predict(payload):
    """Simulate a Prediction function"""
    print("Received:")
    print(payload)
    return

Run App

To initialize the application, use:

$ celery -A my_ml_app worker

Send Message

To send messages for prediction (simulation), let's create a file and call the Celery prediction task.

from my_ml_app import predict

while True:
    msg = input("Enter a message: ")
    if msg == "exit":
        break

    # Call Celery task
    payload = {"message": msg}
    result = predict.delay(payload)

Celery Dashboard - Flower

Celery provides a Dashboard through which we can monitor the status of workers and the work queue. This functionality can be added by installing Flower:

$ python -m pip install flower

Install it, then initialize Flower with:

$ celery -A my_ml_app flower

Access http://0.0.0.0:5555 and browse the available options in the menu.

Horizontal Scaling

Let's check out the horizontal scaling functionality. To do this, we will change the prediction code to simulate a long task.

from celery import Celery
import time

app = Celery("my_ml_app")

# Load the Celery configuration from celeryconfig.py
app.config_from_object("celeryconfig")


@app.task
def predict(payload):
    """Simulate a Slow Prediction function"""

    print("Received:")
    print(payload)

    # Simulate long task
    time.sleep(10)

    print("Ended processing:")
    print(payload)

    return

Let's initialize the application with a limit of up to 10 workers. If Celery identifies the need for more workers, they will be created automatically.

$ celery -A my_ml_app worker --autoscale=10,2

References

• https://docs.celeryq.dev/en/stable/getting-started/index.html