DEV Community: Dhananjay Haridas

Scaling Celery-Based Application in Production

Dhananjay Haridas — Sun, 26 May 2024 14:19:23 +0000

This documentation covers how to scale a Celery-based application for document extraction and comparison using FastAPI, Celery, and Redis. The guide includes steps for task splitting, configuring task dependencies, and scaling individual tasks.

Introduction
Task Definitions
Orchestrating Tasks with Parallel Processing
FastAPI Integration
Scaling Celery Workers
Using Dedicated Queues for Each Task Type
Autoscaling
Distributed Task Execution
Monitoring and Management
Load Balancing and High Availability
Summary

Introduction

This guide provides a detailed explanation of how to scale a Celery-based application that performs document extraction and comparison. It covers breaking down the tasks, orchestrating them for parallel processing, and scaling the application to handle increased loads in a production environment.

Task Definitions

Define the tasks for fetching, extracting, and comparing documents:

# tasks.py

from celery_config import celery_app
import logging

logger = logging.getLogger(__name__)

@celery_app.task
def fetch_documents_task(blob_path):
    try:
        documents = fetch_documents(blob_path)  # Replace with your actual fetch logic
        return documents  # Assume this returns a list of document paths or contents
    except Exception as e:
        logger.error(f"Error fetching documents: {e}")
        raise

@celery_app.task
def extract_data_task(document):
    try:
        extracted_data = extract_data(document)  # Replace with your actual extraction logic
        return extracted_data
    except Exception as e:
        logger.error(f"Error extracting data: {e}")
        raise

@celery_app.task
def compare_data_task(extracted_data_list):
    try:
        comparison_results = compare_data(extracted_data_list)  # Replace with your actual comparison logic
        return comparison_results
    except Exception as e:
        logger.error(f"Error comparing data: {e}")
        raise

Orchestrating Tasks with Parallel Processing

Use a combination of chains and groups to handle dependencies and parallel processing:

# main.py or workflow.py

from celery import chain, group
from tasks import fetch_documents_task, extract_data_task, compare_data_task

def process_documents(blob_path):
    # Step 1: Fetch documents
    fetch_task = fetch_documents_task.s(blob_path)

    # Step 2: Extract data from each document in parallel
    extract_tasks = fetch_task | group(extract_data_task.s(doc) for doc in fetch_task.get())

    # Step 3: Compare the extracted data
    compare_task = compare_data_task.s()

    # Combine the workflow into a single chain
    workflow = chain(fetch_task, extract_tasks, compare_task)
    result = workflow.apply_async()
    return result

FastAPI Integration

Integrate the workflow with a FastAPI endpoint:

# main.py

from fastapi import FastAPI
from workflow import process_documents  # Import your workflow function
from celery_config import celery_app

app = FastAPI()

@app.post("/process/")
async def process_endpoint(blob_path: str):
    result = process_documents(blob_path)
    return {"task_id": result.id}

@app.get("/status/{task_id}")
async def get_status(task_id: str):
    result = celery_app.AsyncResult(task_id)
    if result.state == 'PENDING':
        return {"status": "Pending..."}
    elif result.state == 'SUCCESS':
        return {"status": "Completed", "result": result.result}
    elif result.state == 'FAILURE':
        return {"status": "Failed", "result": str(result.result)}
    else:
        return {"status": result.state}

Scaling Celery Workers

Increasing the Number of Workers

Start multiple Celery worker processes:

celery -A celery_config worker --loglevel=info --concurrency=4

To scale further, start more workers:

celery -A celery_config worker --loglevel=info --concurrency=4
celery -A celery_config worker --loglevel=info --concurrency=4

Distributed Workers

Run workers on different machines by pointing them to the same message broker:

celery -A celery_config worker --loglevel=info --concurrency=4 -Q fetch_queue
celery -A celery_config worker --loglevel=info --concurrency=8 -Q extract_queue
celery -A celery_config worker --loglevel=info --concurrency=2 -Q compare_queue

Using Dedicated Queues for Each Task Type

Defining Queues

Configure Celery to define multiple queues:

# celery_config.py

from celery import Celery

celery_app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

celery_app.conf.task_queues = (
    Queue('fetch_queue', routing_key='fetch.#'),
    Queue('extract_queue', routing_key='extract.#'),
    Queue('compare_queue', routing_key='compare.#'),
)

celery_app.conf.task_routes = {
    'tasks.fetch_documents_task': {'queue': 'fetch_queue', 'routing_key': 'fetch.documents'},
    'tasks.extract_data_task': {'queue': 'extract_queue', 'routing_key': 'extract.data'},
    'tasks.compare_data_task': {'queue': 'compare_queue', 'routing_key': 'compare.data'},
}

Starting Workers for Specific Queues

celery -A celery_config worker --loglevel=info --concurrency=4 -Q fetch_queue
celery -A celery_config worker --loglevel=info --concurrency=8 -Q extract_queue
celery -A celery_config worker --loglevel=info --concurrency=2 -Q compare_queue

Autoscaling

Enable autoscaling to dynamically adjust the number of worker processes:

celery -A celery_config worker --loglevel=info --autoscale=10,3

--autoscale=10,3: Scales between 3 and 10 worker processes based on load.

Distributed Task Execution

Distribute Celery workers across multiple machines:

Example Setup

Machine 1 (Message Broker and Backend):
- Run Redis as your broker and backend.

Machine 2 (Worker Node):

Start Celery workers:

 celery -A celery_config worker --loglevel=info --concurrency=4 -Q fetch_queue

Machine 3 (Worker Node):

Start Celery workers:

 celery -A celery_config worker --loglevel=info --concurrency=8 -Q extract_queue

Machine 4 (Worker Node):

Start Celery workers:

 celery -A celery_config worker --loglevel=info --concurrency=2 -Q compare_queue

Monitoring and Management

Use monitoring tools like Flower, Prometheus, and Grafana to monitor Celery tasks:

Flower

Start Flower to monitor Celery workers:

celery -A celery_config flower

Load Balancing and High Availability

Implement load balancing for high availability and fault tolerance:

Example Load Balancer Setup

Use HAProxy or another load balancer to distribute requests across multiple Redis instances.

Summary

Scale Workers: Increase the number of Celery workers to handle more tasks concurrently.
Dedicated Queues: Use different queues for different types of tasks and scale them independently.
Autoscaling: Enable autoscaling to dynamically adjust the number of worker processes based on load.
Distributed Execution: Distribute workers across multiple machines to improve scalability and fault tolerance.
Monitoring: Use monitoring tools to keep track of the performance and health of your Celery workers.
Load Balancing: Implement load balancing for high availability and fault tolerance.

By following these strategies, you can effectively scale your Celery-based application to handle increased loads and ensure reliable task execution in a production environment.

MongoDB Shell Commands - A Comprehensive Guide

Dhananjay Haridas — Sat, 13 Apr 2024 22:04:49 +0000

Step 1: Start MongoDB Shell

Open your terminal.
Run the mongo command to start the MongoDB shell:

mongo

Step 2: Switch to a Database

To switch to a specific database, use the use command followed by the name of the database:

use mydatabase

Replace mydatabase with the name of your database.

Step 3: Show Databases

To show all databases available on the MongoDB server, run the following command:

show databases

This command will list all databases, including empty databases.

Step 4: Show Collections

After switching to a database, you can list all collections within that database using the show collections command:

show collections

This command will display all collections present in the current database.

Step 5: Show Data from a Collection

To display all documents (data) in a collection, use the db.collectionName.find() method. Replace collectionName with the name of your collection:

db.collectionName.find()

For example, to display all documents in a collection named users:

db.users.find()

You can also apply query criteria to filter the results. For example, to find documents where the age field is greater than 18:

db.collectionName.find({ age: { $gt: 18 } })

Step 6: Insert Documents into a Collection

To insert documents into a collection, use the db.collectionName.insertOne() or db.collectionName.insertMany() methods.

For example, to insert a single document into a collection named users:

db.users.insertOne({ name: "John", age: 30 })

To insert multiple documents into a collection:

db.users.insertMany([
    { name: "Alice", age: 25 },
    { name: "Bob", age: 35 }
])

Step 7: Update Documents in a Collection

To update documents in a collection, use the db.collectionName.updateOne() or db.collectionName.updateMany() methods.

For example, to update a single document in the users collection:

db.users.updateOne({ name: "John" }, { $set: { age: 31 } })

To update multiple documents:

db.users.updateMany({ age: { $lt: 30 } }, { $set: { category: "Young" } })

Step 8: Delete Documents from a Collection

To delete documents from a collection, use the db.collectionName.deleteOne() or db.collectionName.deleteMany() methods.

For example, to delete a single document from the users collection:

db.users.deleteOne({ name: "John" })

To delete multiple documents:

db.users.deleteMany({ category: "Young" })

Step 9: Exit MongoDB Shell

To exit the MongoDB shell, type exit and press Enter:

exit

These commands cover the basic CRUD operations (Create, Read, Update, Delete) in MongoDB from the shell. Adjust the database, collection names, and query criteria as needed for your specific use case.

MongoDB Installation Guide for Windows Subsystem for Linux (WSL)

Dhananjay Haridas — Sat, 13 Apr 2024 21:55:17 +0000

This guide will assist you in installing MongoDB using the Windows Subsystem for Linux (WSL) via the Command Line.

Install MongoDB - Community Edition

Launch a terminal (the Ubuntu app) and navigate to the root of the Ubuntu File System by typing cd ~.
Copy and paste the following command into the terminal to import the MongoDB public GPG Key:

   sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 2930ADAE8CAF5059EE73BB4B58712A2291FA4AD5

Next, add the MongoDB package to your sources list by pasting this line into the terminal:

   echo "deb [ arch=amd64,arm64 ] https://repo.mongodb.org/apt/ubuntu xenial/mongodb-org/3.6 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-3.6.list

Note: Ensure you are on Ubuntu Xenial. We'll update this for Zesty LTS when available.

Refresh your local package database with the command:

   sudo apt-get update

Install MongoDB by running the following command:

   sudo apt-get install -y mongodb-org

This will install the latest stable version (MongoDB 3.6 at the time of writing). Refer to the provided link if you wish to install a different version.

Resolve Unmet Dependencies Error

If you encounter an error regarding unmet dependencies, execute the following steps:

Download libssl1.1_1.1.1f-1ubuntu2_amd64.deb:

   wget http://archive.ubuntu.com/ubuntu/pool/main/o/openssl/libssl1.1_1.1.1f-1ubuntu2_amd64.deb

Install the downloaded package using dpkg:

   sudo dpkg -i libssl1.1_1.1.1f-1ubuntu2_amd64.deb

Create Data and DB Directories

Ensure you are still in the root of the Ubuntu FS by typing cd ~. Confirm with pwd, which should display /home/<user>/.
Create the necessary directories by typing:

   sudo mkdir -p data/db

This command creates a data directory with a db subdirectory. The -p flag ensures the parent directory is created if it doesn't exist.

Start mongod Server and mongo Shell

Open a new terminal window and run the following command to start the MongoDB server:

   sudo mongod --dbpath ~/data/db

You should see status messages, with the last line indicating that the server is waiting for connections on port 27017.

In another WSL window, type mongo to launch the MongoDB shell.
You'll see a connection notification in the first terminal window and the prompt in the new window should change to > indicating you're in the mongo shell.
Refer to this cheat sheet for guidance on using the mongo shell.
To exit the shell, press Ctrl + C. You'll receive a confirmation message before returning to the command line.

Troubleshooting

For further assistance, consult the MongoDB Install Docs or feel free to reach out directly via message or email.

credis -
github
stackoverflow

Windows Subsystem for Linux (WSL) Installation and Configuration Guide

Dhananjay Haridas — Sat, 06 Apr 2024 08:37:39 +0000

This comprehensive guide offers detailed instructions on setting up and configuring Ubuntu within the Windows Subsystem for Linux (WSL). WSL is a compatibility layer provided by Microsoft that enables users to run native Linux command-line tools directly on Windows 10 and Windows 11. By integrating a Linux kernel interface, WSL bridges the gap between the Windows and Linux ecosystems, providing users with the flexibility to leverage the strengths of both operating systems without the need for dual-booting or virtual machines.

Why Use WSL?

Seamless Integration: With WSL, users can seamlessly integrate Linux tools and utilities into their Windows workflow, enhancing productivity and development capabilities.
Compatibility: WSL ensures compatibility with a wide range of Linux software, allowing users to run popular applications, programming languages, and development environments natively on Windows.
Resource Efficiency: Unlike traditional virtual machines, WSL offers lightweight resource utilization, minimizing overhead and maximizing system performance.
Development Environment: WSL provides developers with a familiar Linux environment for software development, enabling them to utilize command-line tools, package managers, and development frameworks seamlessly.

Installing WSL (Windows Subsystem for Linux)

Open PowerShell as Administrator: Right-click the Start menu, then select "Windows PowerShell (Admin)".
Install WSL: Run the following command in PowerShell:

   wsl --install

This command will enable the WSL feature on your system.

Switching to WSL 2

Set Default WSL Version to 2: After installing WSL, switch to WSL 2 by running:

   wsl --set-default-version 2

This ensures that any new Linux distributions installed will use WSL 2.

Managing Linux Distributions

List Available Linux Distributions: To see all available Linux distributions in the Microsoft Store, use:

   wsl --list --online

This command will display a list of Linux distributions that you can install.

Install Ubuntu: If you choose Ubuntu, you can install it directly using:

   wsl --install -d Ubuntu

This command will download and set up Ubuntu on your system.

Check Installed Distributions: To see which Linux distributions are installed and their versions, run:

   wsl -l -v

This will list all installed distributions along with their respective versions.

Configuring Ubuntu

Set Default Distribution: If Ubuntu is not your default distribution, you can set it using:

   wsl --setdefault Ubuntu

This will make Ubuntu the default distribution to launch when using WSL.

Configure Ubuntu: After setting Ubuntu as default, update its package lists by running:

   sudo apt update

This ensures that you have the latest package information.

This Markdown document provides a step-by-step guide for installing, managing, and configuring WSL with Ubuntu on your Windows system.