Image for post Celery para Tareas Asíncronas y Distribuidas en Python: Escalando tus Procesos de Datos e IA

Celery para Tareas Asíncronas y Distribuidas en Python: Escalando tus Procesos de Datos e IA

Escrito por Sergio Márquez

Contexto del Problema

Como desarrolladores, a menudo nos enfrentamos a la necesidad de ejecutar operaciones que consumen mucho tiempo o recursos. Imagina que estás construyendo una API que permite a los usuarios subir imágenes para aplicarles un filtro complejo de IA, o una aplicación que genera reportes extensos basados en grandes volúmenes de datos. Si estas operaciones se ejecutan directamente en el hilo principal de tu aplicación web, el usuario experimentará un retraso significativo, o peor aún, la aplicación podría bloquearse, resultando en una mala experiencia de usuario y una baja escalabilidad.

Aquí es donde entran en juego las tareas asíncronas y distribuidas. En lugar de ejecutar estas operaciones pesadas de forma síncrona, podemos "delegarlas" a otro proceso o incluso a otra máquina. Esto libera a nuestra aplicación principal para que pueda seguir respondiendo rápidamente a otras solicitudes, mientras las tareas intensivas se procesan en segundo plano. En el ámbito de la Inteligencia Artificial, esto es crucial para:

  • Procesamiento de datos: Limpieza, transformación o enriquecimiento de grandes datasets antes del entrenamiento o la inferencia.
  • Inferencia en lotes: Aplicar un modelo de IA a un gran número de entradas de una sola vez, sin bloquear la API de inferencia en tiempo real.
  • Entrenamiento de modelos: Ejecutar sesiones de entrenamiento de modelos que pueden durar minutos u horas.
  • Generación de reportes o análisis complejos: Procesar datos y generar visualizaciones o documentos que requieren tiempo.

Celery es una de las herramientas más populares y robustas en el ecosistema Python para gestionar este tipo de tareas. Nos permite definir funciones de Python como "tareas" que pueden ser ejecutadas de forma asíncrona por uno o varios "workers" distribuidos, utilizando una "cola de mensajes" para coordinar el trabajo.

Conceptos Clave

Para entender cómo funciona Celery, es fundamental familiarizarse con algunos conceptos:

Cola de Tareas (Task Queue)

Una cola de tareas es un mecanismo que permite a una aplicación (el cliente) enviar mensajes que representan unidades de trabajo (tareas) a otra aplicación (el worker) para que las procese. La cola actúa como un búfer, almacenando las tareas hasta que un worker esté disponible para ejecutarlas. Esto desacopla el proceso de envío de tareas del proceso de ejecución, mejorando la resiliencia y la escalabilidad.

Broker de Mensajes (Message Broker)

El broker de mensajes es el corazón de la comunicación en Celery. Es una pieza de software que implementa la cola de tareas. Cuando un cliente envía una tarea, en realidad la envía al broker. Cuando un worker está listo para procesar una tarea, la solicita al broker. Celery soporta varios brokers, siendo los más comunes:

  • Redis: Un almacén de datos en memoria, muy rápido y popular para colas de mensajes simples.
  • RabbitMQ: Un broker de mensajes más robusto y con más características, ideal para entornos de producción con requisitos de alta disponibilidad y durabilidad.

Para este artículo, usaremos Redis por su facilidad de configuración y uso.

Worker

Un worker es un proceso (o un conjunto de procesos) que se encarga de escuchar el broker de mensajes, tomar tareas de la cola y ejecutarlas. Puedes tener uno o varios workers ejecutándose en la misma máquina o distribuidos en diferentes servidores. Esto permite escalar horizontalmente tu capacidad de procesamiento de tareas.

Cliente (Client)

El cliente es la parte de tu aplicación que "produce" tareas, es decir, que las envía al broker para que sean procesadas. Puede ser una API web, un script de línea de comandos, o cualquier otra aplicación Python.

Tarea (Task)

En Celery, una tarea es una función de Python regular que ha sido decorada con @celery_app.task. Esta decoración le indica a Celery que esta función puede ser enviada a la cola y ejecutada por un worker. Las tareas deben ser idempotentes (ejecutarlas múltiples veces produce el mismo resultado) y no deben depender del estado de la aplicación cliente, ya que se ejecutarán en un entorno separado.

Resultado (Result Backend)

Opcionalmente, Celery puede almacenar el resultado de una tarea una vez que ha sido completada. Esto se hace configurando un "backend de resultados". Al igual que el broker, puede ser Redis, una base de datos (como PostgreSQL), o incluso el propio RabbitMQ. El backend de resultados permite al cliente consultar el estado de una tarea (pendiente, en progreso, completada, fallida) y recuperar su valor de retorno.

Implementación Paso a Paso

Vamos a configurar un entorno básico con Celery y Redis.

1. Configuración del Entorno

Primero, asegúrate de tener Python y pip instalados. También necesitarás un servidor Redis ejecutándose. Si no lo tienes, puedes instalarlo fácilmente en la mayoría de los sistemas operativos o usar Docker.

Crea un nuevo directorio para tu proyecto y un entorno virtual:


mkdir celery_project
cd celery_project
python -m venv venv
source venv/bin/activate  # En Windows: venv\Scripts\activate

Instala las dependencias necesarias:


pip install celery redis

2. Creación de la Aplicación Celery

Crea un archivo llamado celery_app.py. Aquí definiremos nuestra instancia de Celery y configuraremos el broker de mensajes y el backend de resultados.


import os
from celery import Celery

# Configura la URL del broker de Redis desde una variable de entorno
# Por defecto, asume Redis en localhost:6379
REDIS_BROKER_URL = os.getenv('REDIS_BROKER_URL', 'redis://localhost:6379/0')
REDIS_RESULT_BACKEND = os.getenv('REDIS_RESULT_BACKEND', 'redis://localhost:6379/1')

celery_app = Celery(
    'my_celery_app',
    broker=REDIS_BROKER_URL,
    backend=REDIS_RESULT_BACKEND
)

# Opcional: Configuración adicional
celery_app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
)

# Importa las tareas aquí para que Celery las descubra
# from . import tasks # Si las tareas estuvieran en un archivo tasks.py

Hemos usado variables de entorno (REDIS_BROKER_URL, REDIS_RESULT_BACKEND) para la configuración de Redis. Esto es una buena práctica para mantener las credenciales y configuraciones sensibles fuera del código fuente.

3. Definición de una Tarea Simple

Ahora, definamos una tarea que simule un procesamiento largo. Crea un archivo tasks.py:


import time
from celery_app import celery_app

@celery_app.task
def add(x, y):
    print(f"Ejecutando tarea add({x}, {y})...")
    time.sleep(5) # Simula un trabajo pesado de 5 segundos
    result = x + y
    print(f"Tarea add({x}, {y}) completada. Resultado: {result}")
    return result

@celery_app.task
def multiply(x, y):
    print(f"Ejecutando tarea multiply({x}, {y})...")
    time.sleep(3) # Simula un trabajo pesado de 3 segundos
    result = x * y
    print(f"Tarea multiply({x}, {y}) completada. Resultado: {result}")
    return result

Asegúrate de importar celery_app desde tu archivo celery_app.py. El decorador @celery_app.task convierte la función Python en una tarea de Celery.

4. Envío de Tareas

Para enviar tareas, crearemos un script cliente. Crea producer.py:


from tasks import add, multiply
import time

print("Enviando tareas...")

# Envía la tarea 'add' y obtiene un objeto AsyncResult
result_add = add.delay(10, 20)
print(f"Tarea 'add' enviada. ID: {result_add.id}")

# Envía la tarea 'multiply' con apply_async para más control
result_multiply = multiply.apply_async(args=[5, 6])
print(f"Tarea 'multiply' enviada. ID: {result_multiply.id}")

print("Esperando resultados (esto bloqueará hasta que las tareas terminen)...")

# Puedes consultar el estado y el resultado de la tarea
# .get() bloqueará hasta que la tarea termine
print(f"Resultado de 'add': {result_add.get(timeout=10)}")
print(f"Resultado de 'multiply': {result_multiply.get(timeout=10)}")

print("Todas las tareas completadas.")

.delay() es un atajo para .apply_async() con argumentos posicionales. Ambos métodos devuelven un objeto AsyncResult, que nos permite consultar el estado y el resultado de la tarea.

5. Inicio del Worker

Para que las tareas se ejecuten, necesitas iniciar un worker de Celery. Abre una nueva terminal en el directorio de tu proyecto y ejecuta:


# Asegúrate de que tu entorno virtual esté activado
# export REDIS_BROKER_URL='redis://localhost:6379/0' # Si no usas el valor por defecto
# export REDIS_RESULT_BACKEND='redis://localhost:6379/1' # Si no usas el valor por defecto
celery -A tasks worker --loglevel=info

El argumento -A tasks le dice a Celery que busque tareas en el módulo tasks.py. --loglevel=info mostrará información útil en la consola del worker.

6. Ejecución de Prueba

Ahora, en tu primera terminal (donde ejecutaste source venv/bin/activate), ejecuta el productor:


python producer.py

Observa cómo el productor envía las tareas y luego espera. En la terminal del worker, verás cómo las tareas son recibidas y ejecutadas. El productor imprimirá los resultados una vez que las tareas hayan finalizado.

Mini Proyecto / Aplicación Sencilla: Procesamiento de Imágenes Asíncrono con FastAPI y Celery

Vamos a construir una pequeña API con FastAPI que permita subir una imagen, y en lugar de procesarla inmediatamente, delegará la tarea a Celery. El cliente podrá consultar el estado del procesamiento.

1. Instalación de Dependencias Adicionales


pip install fastapi uvicorn Pillow

2. Modificación de celery_app.py

Asegúrate de que celery_app.py esté configurado como antes.

3. Creación de Tareas de Procesamiento de Imagen (tasks.py)

Modifica tasks.py para incluir una tarea de procesamiento de imagen. Usaremos la librería Pillow para una manipulación básica.


import time
import base64
from io import BytesIO
from PIL import Image
from celery_app import celery_app

@celery_app.task
def process_image_task(image_base64: str, task_id: str):
    print(f"[Worker] Iniciando procesamiento de imagen para la tarea {task_id}...")
    try:
        # Decodificar la imagen de base64
        image_bytes = base64.b64decode(image_base64)
        img = Image.open(BytesIO(image_bytes))

        # Simular un procesamiento pesado (ej. aplicar un filtro, redimensionar)
        time.sleep(7) # Simula 7 segundos de trabajo de IA
        
        # Ejemplo de procesamiento: convertir a escala de grises
        img_processed = img.convert("L") 

        # Guardar la imagen procesada en un buffer y codificarla de nuevo a base64
        buffered = BytesIO()
        img_processed.save(buffered, format="PNG")
        processed_image_base64 = base64.b64encode(buffered.getvalue()).decode('utf-8')

        print(f"[Worker] Procesamiento de imagen para la tarea {task_id} completado.")
        return {"status": "completed", "processed_image": processed_image_base64}
    except Exception as e:
        print(f"[Worker] Error procesando imagen para la tarea {task_id}: {e}")
        return {"status": "failed", "error": str(e)}

# Mantén las tareas add y multiply si quieres, o elimínalas para este ejemplo
@celery_app.task
def add(x, y):
    print(f"Ejecutando tarea add({x}, {y})...")
    time.sleep(5)
    result = x + y
    print(f"Tarea add({x}, {y}) completada. Resultado: {result}")
    return result

La tarea process_image_task recibe la imagen codificada en base64 para evitar problemas de serialización de objetos complejos de Pillow directamente en Celery. Devuelve la imagen procesada también en base64.

4. Creación de la API FastAPI (main.py)


from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import JSONResponse
from celery.result import AsyncResult
import base64
from io import BytesIO
from PIL import Image

from tasks import process_image_task # Importa la tarea

app = FastAPI(title="API de Procesamiento de Imágenes con Celery")

@app.post("/process-image/")
async def process_image(file: UploadFile = File(...)):
    if not file.content_type.startswith('image/'):
        raise HTTPException(status_code=400, detail="El archivo debe ser una imagen.")

    try:
        image_bytes = await file.read()
        # Opcional: Validar que es una imagen válida antes de enviar a Celery
        Image.open(BytesIO(image_bytes)).verify()

        # Codificar la imagen a base64 para pasarla a Celery
        image_base64 = base64.b64encode(image_bytes).decode('utf-8')
        
        # Enviar la tarea a Celery
        task = process_image_task.delay(image_base64, "unique_image_id_" + str(hash(image_base64)))
        
        return JSONResponse({
            "message": "Procesamiento de imagen iniciado",
            "task_id": task.id,
            "status_url": f"/task-status/{task.id}"
        })
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error al iniciar el procesamiento: {e}")

@app.get("/task-status/{task_id}")
async def get_task_status(task_id: str):
    task_result = AsyncResult(task_id, app=process_image_task.app)
    
    if task_result.ready(): # La tarea ha terminado (éxito o fallo)
        result_data = task_result.get()
        if result_data.get("status") == "completed":
            return JSONResponse({
                "task_id": task_id,
                "status": "completed",
                "result": result_data.get("processed_image"), # La imagen procesada en base64
                "message": "Imagen procesada exitosamente."
            })
        else:
            return JSONResponse({
                "task_id": task_id,
                "status": "failed",
                "error": result_data.get("error", "Error desconocido"),
                "message": "La tarea falló."
            }, status_code=500)
    else:
        return JSONResponse({
            "task_id": task_id,
            "status": task_result.status,
            "message": "Tarea en progreso o pendiente."
        })

@app.get("/health")
async def health_check():
    return {"status": "ok", "message": "API is running"}

5. Ejecución del Mini Proyecto

Abre tres terminales:

  1. Terminal 1 (Redis): Asegúrate de que Redis esté corriendo. Si lo instalaste localmente, simplemente inicia el servicio. Si usas Docker:

    
docker run --name some-redis -p 6379:6379 -d redis

  2. Terminal 2 (Worker de Celery): Inicia el worker. Asegúrate de que el entorno virtual esté activado.

    
celery -A tasks worker --loglevel=info

  3. Terminal 3 (API FastAPI): Inicia la API.

    
uvicorn main:app --reload

6. Prueba de la Aplicación

Puedes usar una herramienta como Postman, Insomnia o curl para probar la API. Necesitarás una imagen de prueba (ej. test_image.png).


# Ejemplo usando curl para enviar una imagen
# Asegúrate de tener una imagen llamada 'test_image.png' en el mismo directorio

curl -X POST "http://127.0.0.1:8000/process-image/" \
  -H "accept: application/json" \
  -H "Content-Type: multipart/form-data" \
  -F "file=@test_image.png;type=image/png"

La respuesta inicial de la API te dará un task_id y una status_url. Por ejemplo:


{
  "message": "Procesamiento de imagen iniciado",
  "task_id": "algún-id-único",
  "status_url": "/task-status/algún-id-único"
}

Luego, puedes consultar el estado de la tarea:


curl -X GET "http://127.0.0.1:8000/task-status/algún-id-único" \
  -H "accept: application/json"

Inicialmente, verás un estado como PENDING o STARTED. Después de unos 7 segundos (el time.sleep(7) en la tarea), si vuelves a consultar, deberías ver COMPLETED y la imagen procesada en base64.

Errores Comunes y Depuración

  • Worker no se inicia o no encuentra las tareas:

    • Asegúrate de que el argumento -A en celery -A tasks worker apunte correctamente al módulo donde están definidas tus tareas (ej. tasks para tasks.py).
    • Verifica que el entorno virtual esté activado y que Celery esté instalado.
    • Revisa los logs del worker para ver mensajes de error.

  • Problemas de conexión con el broker (Redis):

    • Asegúrate de que Redis esté ejecutándose y sea accesible desde la máquina donde se ejecuta el worker y el cliente.
    • Verifica que REDIS_BROKER_URL y REDIS_RESULT_BACKEND estén configurados correctamente (host, puerto, número de base de datos).
    • Un error común es que el worker no pueda conectarse a Redis, lo que se manifestará en los logs del worker.

  • Serialización de argumentos de tareas:

    • Celery necesita serializar los argumentos de las tareas para enviarlos a través del broker. Por defecto, usa json. Esto significa que solo puedes pasar tipos de datos que sean serializables en JSON (números, cadenas, listas, diccionarios, booleanos, None).
    • Si intentas pasar objetos complejos (como objetos de Pillow, conexiones a bases de datos, etc.), Celery fallará. La solución es serializar estos objetos manualmente (ej. a base64 como hicimos con las imágenes) antes de pasarlos a la tarea.

  • Manejo de excepciones dentro de las tareas:

    • Las excepciones no manejadas dentro de una tarea harán que la tarea falle. Es una buena práctica envolver el código de la tarea en bloques try-except para capturar errores y devolver un estado de fallo o registrar el error adecuadamente.
    • El backend de resultados almacenará el traceback de la excepción, lo que es útil para la depuración.

  • Versiones de Celery y dependencias:

    • Celery es una librería madura, pero las versiones pueden tener cambios sutiles. Siempre consulta la documentación oficial para la versión específica que estás usando.
    • Asegúrate de que las versiones de celery y redis (o amqp para RabbitMQ) sean compatibles.

Aprendizaje Futuro / Próximos Pasos

Has dado los primeros pasos con Celery, pero hay mucho más que explorar para construir sistemas robustos y escalables:

  • Monitoreo con Flower: Flower es una herramienta de monitoreo en tiempo real para Celery. Proporciona una interfaz web para ver el estado de las tareas, los workers, y estadísticas. Es invaluable para la depuración y la observación en producción.

    
pip install flower
celery -A tasks flower --port=5555

    Luego, accede a http://localhost:5555 en tu navegador.

  • Encadenamiento de Tareas (Chains, Chords, Groups): Celery permite definir flujos de trabajo complejos donde el resultado de una tarea alimenta a la siguiente, o donde múltiples tareas se ejecutan en paralelo y sus resultados se combinan. Esto es fundamental para pipelines de datos e IA.

  • Tareas Periódicas (Celery Beat): Si necesitas ejecutar tareas a intervalos regulares (ej. cada hora, cada día), Celery Beat es tu herramienta. Funciona como un planificador de tareas (cron) pero integrado con Celery.

  • Escalado de Workers: Aprende a configurar múltiples workers, a distribuirlos en diferentes máquinas y a usar herramientas como Docker y Kubernetes para orquestarlos y escalarlos automáticamente según la carga.

  • Uso de RabbitMQ: Para entornos de producción, RabbitMQ es a menudo la opción preferida como broker de mensajes debido a sus características de durabilidad, enrutamiento avanzado y alta disponibilidad.

  • Integración con Docker: Contenerizar tu aplicación FastAPI, Celery worker y Redis en Docker Compose es un paso crucial para simplificar el despliegue y asegurar la consistencia del entorno.

Celery es una herramienta poderosa que te permitirá construir aplicaciones Python más reactivas, escalables y resilientes, especialmente cuando trabajas con procesos intensivos en datos o IA. ¡Experimenta y sigue construyendo!

Volver al blog