Pipeline de Datos y API con Caché Inteligente - IMDB Movies

Objetivo del Proyecto

Desarrollar una solución integral de backend que demuestre la migración de datos a gran escala desde Kaggle hacia Azure, exponiendo un catálogo de películas IMDB a través de una API segura y optimizada con:

• Sistema de caché inteligente
• Autenticación robusta
• Monitoreo en tiempo real

Arquitectura del Sistema

La solución implementa una arquitectura de microservicios distribuida en Azure con los siguientes componentes:

Componentes Principales

Fase 1: Pipeline ETL y Migración de Datos

Dataset: IMDB Movies Dataset (Kaggle)

{
  "series_title": "The Shawshank Redemption",
  "released_year": 1994,
  "certificate": "A",
  "runtime": "142 min",
  "genre": "Drama",
  "imdb_rating": 9.3,
  "overview": "Two imprisoned...",
  "meta_score": 80,
  "director": "Frank Darabont",
  "star1": "Tim Robbins",
  "star2": "Morgan Freeman",
  "star3": "Bob Gunton",
  "star4": "William Sadler",
  "no_of_votes": 2343110,
  "gross": "16,000,000"
}

Arquitectura ETL

graph LR
    A[Kaggle Dataset] --> B[Azure Blob Storage]
    B --> C[Azure Data Factory]
    C --> D[Data Transformation]
    D --> E[Azure SQL Database]
    E --> F[FastAPI Application]

Pipeline de Transformación

• Limpieza de datos nulos y duplicados
• Normalización de géneros y certificaciones
• Conversión de tipos de datos optimizada
• Indexación por género, año y director

Fase 2: API REST con Autenticación

Modelo de Datos

from pydantic import BaseModel, Field
from typing import Optional, List

class MovieCatalog(BaseModel):
    id: Optional[int] = None
    poster_link: Optional[str] = None
    series_title: str
    released_year: int
    certificate: Optional[str] = None
    runtime: Optional[str] = None
    genre: str
    imdb_rating: float
    overview: Optional[str] = None
    meta_score: Optional[int] = None
    director: str
    stars: List[str] = Field(default_factory=list)
    no_of_votes: Optional[int] = None
    gross: Optional[str] = None

Endpoints Implementados

Decorador para validación de administradores

@validateadmin
async def create_new_movie(request: Request, movie_data: MovieCatalog):
    new_movie = await create_movie(movie_data)
    return new_movie

Fase 3: Sistema de Caché Inteligente con Redis

Generación Dinámica de Claves

cache_key = "movies:catalog"
if genre:
    cache_key += f":genre={genre.lower()}"
if year:
    cache_key += f":year={year}"
if director:
    cache_key += f":director={director.lower()}"
if min_rating:
    cache_key += f":min_rating={min_rating}"

Ejemplos de claves generadas:

• movies:catalog:genre=action:year=2020
• movies:catalog:director=christopher nolan:min_rating=8.0

⚡ Performance del Caché

Invalidación Inteligente

async def create_movie(movie_data: MovieCatalog):
    await execute_query_json(insert_query, params)
    
    keys_to_delete = ["movies:catalog"]
    for genre in movie_data.genre.split(", "):
        keys_to_delete.append(f"movies:catalog:genre={genre.lower()}")
    
    keys_to_delete.extend([
        f"movies:catalog:year={movie_data.released_year}",
        f"movies:catalog:director={movie_data.director.lower()}"
    ])
    
    for key in set(keys_to_delete):
        delete_cache(redis_client, key)

Fase 4: Monitoreo y Telemetría

Integración con Application Insights

def setup_simple_telemetry():
    connection_string = os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")
    if connection_string:
        configure_azure_monitor(connection_string=connection_string)
        return True
    return False

instrument_fastapi_app(app)

Métricas Monitoreadas

• Latencia de Endpoints: P50, P95, P99
• Cache Hit Ratio: Efectividad del caché Redis
• Errores de Autenticación: Intentos fallidos
• Database Performance: Tiempo de consultas SQL
• Memory Usage: Uso de memoria del contenedor

Fase 5: Containerización y Despliegue

Dockerfile Optimizado

FROM python:3.12-slim

WORKDIR /app

RUN apt-get update && \
    apt-get install -y curl gcc g++ && \
    curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \
    ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev && \
    apt-get clean

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Arquitectura de Despliegue

graph TB
    A[Azure Container Registry] --> B[Azure Container Apps]
    B --> C[Azure SQL Database]
    B --> D[Redis Cache]
    B --> E[Application Insights]
    F[Azure Front Door] --> B
    G[Firebase Auth] --> B

Resultados y Métricas

Performance Benchmarks (1000 requests concurrentes)

Arquitectura de Alta Disponibilidad

• Redundancia: Multi-AZ deployment
• Auto-scaling: CPU y memoria
• Health Checks: /health cada 30s
• Circuit Breaker + Retry Logic

Tecnologías y Herramientas

Backend Stack

• FastAPI, Pydantic, SQLAlchemy, Redis

Infraestructura Cloud

• Azure Data Factory, Azure SQL Database, Azure Container Apps, Application Insights

DevOps y Seguridad

• Docker, Firebase Auth, Azure Key Vault, GitHub Actions

Casos de Uso Implementados

1. Búsqueda Avanzada de Catálogo

GET /catalog?genre=Action&year=2020&min_rating=7.5
Authorization: Bearer <jwt_token>

2. Creación de Contenido (Admin)

POST /catalog
Authorization: Bearer <admin_jwt_token>
Content-Type: application/json

{
  "series_title": "New Blockbuster",
  "released_year": 2024,
  "genre": "Action, Sci-Fi",
  "imdb_rating": 8.2,
  "director": "Christopher Nolan"
}

3. Invalidación Automática de Caché

Al crear una nueva película de género "Action" en 2024, el sistema automáticamente invalida:

• movies:catalog
• movies:catalog:genre=action
• movies:catalog:year=2024

Conclusiones

Este proyecto demuestra la implementación de una arquitectura de microservicios moderna y escalable, integrando:

• ETL Robusto: Migración automatizada de 50K+ registros
• API Performante: Sub-50ms de latencia promedio
• Caché Inteligente: 98%+ de mejora en performance
• Seguridad Enterprise: Autenticación JWT + roles
• Observabilidad Completa: Monitoreo en tiempo real
• DevOps Maduro: CI/CD + containerización

La solución está preparada para escalar horizontalmente y manejar millones de requests diarios manteniendo alta disponibilidad y performance optimizada.