Cours 16 : Projet Fil Rouge - DataPipeline

1. Vision et Architecture

1.1 Vue d'ensemble

DataPipeline est une plateforme SaaS de traitement de donnees qui permet :

• Ingestion depuis multiples sources (API, fichiers, streams)
• Transformation avec Polars/Pandas
• Machine Learning avec scikit-learn et LangChain
• Monitoring avec Prometheus et Grafana
• API REST avec FastAPI

1.2 Architecture globale

Diagramme en cours de génération...

2. Backend FastAPI

2.1 Configuration

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    app_name: str = "DataPipeline"
    debug: bool = False
    database_url: str = "postgresql+asyncpg://user:pass@localhost/db"
    redis_url: str = "redis://localhost:6379/0"
    celery_broker_url: str = "redis://localhost:6379/1"
    mlflow_tracking_uri: str = "http://localhost:5000"
    secret_key: str
    environment: str = "development"

    class Config:
        env_file = ".env"

@lru_cache()
def get_settings() -> Settings:
    return Settings()

2.2 Application principale

from fastapi import FastAPI
from contextlib import asynccontextmanager
from prometheus_fastapi_instrumentator import Instrumentator

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await database.connect()
    yield
    # Shutdown
    await database.disconnect()

app = FastAPI(title="DataPipeline API", version="1.0.0", lifespan=lifespan)

# Routers
app.include_router(pipelines.router, prefix="/api/v1/pipelines")
app.include_router(models.router, prefix="/api/v1/models")
app.include_router(health.router, prefix="/api/v1")

# Monitoring
Instrumentator().instrument(app).expose(app)

# Middleware
app.add_middleware(LoggingMiddleware)
app.add_middleware(RateLimitMiddleware)

2.3 Schemas Pydantic

from pydantic import BaseModel, Field
from uuid import UUID, uuid4
from datetime import datetime
from enum import Enum

class PipelineStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

class PipelineCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str = ""
    source_type: str = Field(..., pattern="^(api|file|stream)$")
    config: dict = {}

class PipelineResponse(BaseModel):
    id: UUID
    name: str
    status: PipelineStatus
    created_at: datetime
    updated_at: datetime

class PipelineRun(BaseModel):
    pipeline_id: UUID
    started_at: datetime
    completed_at: datetime | None = None
    status: PipelineStatus
    rows_processed: int = 0
    error: str | None = None

3. Workers Celery

3.1 Configuration

from celery import Celery
from src.core.config import get_settings

settings = get_settings()

celery_app = Celery(
    "datapipeline",
    broker=settings.celery_broker_url,
    backend=settings.redis_url,
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    task_soft_time_limit=3600,
    task_time_limit=7200,
    worker_max_tasks_per_child=100,
    task_acks_late=True,
    worker_prefetch_multiplier=1,
)

3.2 Tasks

from celery import shared_task
import polars as pl
import mlflow
from src.services.data_service import fetch_data, transform_data
from src.services.storage_service import store_results

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def run_pipeline(self, pipeline_id: str):
    try:
        pipeline = get_pipeline_from_db(pipeline_id)
        update_status(pipeline_id, "running")

        raw_data = fetch_data(pipeline.source_config)
        df = pl.from_pandas(raw_data)

        for step in pipeline.transform_steps:
            df = execute_transform(df, step)

        result = store_results(pipeline_id, df)
        update_status(pipeline_id, "completed")

        return {
            "pipeline_id": pipeline_id,
            "status": "completed",
            "rows": len(df),
            "result_path": result,
        }
    except Exception as exc:
        update_status(pipeline_id, "failed", str(exc))
        raise self.retry(exc=exc)

@shared_task
def train_model(dataset_id: str, model_config: dict):
    with mlflow.start_run() as run:
        mlflow.log_params(model_config)
        X, y = load_training_data(dataset_id)
        model = train_model(X, y, model_config)
        metrics = evaluate(model, X, y)
        mlflow.log_metrics(metrics)
        mlflow.sklearn.log_model(model, "model")
        mlflow.register_model(f"runs:/{run.info.run_id}/model", "production")
        return {"run_id": run.info.run_id, "metrics": metrics}

4. ML Pipeline

4.1 Service ML

import mlflow
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.metrics import accuracy_score, f1_score, classification_report

class MLService:
    def __init__(self):
        mlflow.set_tracking_uri(get_settings().mlflow_tracking_uri)
        mlflow.set_experiment("datapipeline-ml")

    def train(self, dataset, target_column, model_config):
        X = dataset.drop(columns=[target_column])
        y = dataset[target_column]
        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

        with mlflow.start_run():
            pipeline = Pipeline([
                ("scaler", StandardScaler()),
                ("classifier", RandomForestClassifier(
                    n_estimators=model_config.get("n_estimators", 100),
                    max_depth=model_config.get("max_depth"),
                )),
            ])
            pipeline.fit(X_train, y_train)
            y_pred = pipeline.predict(X_test)

            metrics = {
                "accuracy": accuracy_score(y_test, y_pred),
                "f1": f1_score(y_test, y_pred, average="weighted"),
            }
            mlflow.log_params(model_config)
            mlflow.log_metrics(metrics)
            mlflow.sklearn.log_model(pipeline, "model")

            return {"run_id": mlflow.active_run().info.run_id, **metrics}

    def predict(self, model_uri: str, data):
        model = mlflow.pyfunc.load_model(model_uri)
        return model.predict(data).tolist()

5. Monitoring

5.1 Metriques Prometheus

from prometheus_client import Counter, Histogram, Gauge
import time

REQUESTS_TOTAL = Counter(
    "api_requests_total",
    "Total API requests",
    ["method", "endpoint", "status"],
)
REQUESTS_LATENCY = Histogram(
    "api_request_latency_seconds",
    "Request latency in seconds",
    ["method", "endpoint"],
)
ACTIVE_PIPELINES = Gauge(
    "active_pipelines",
    "Number of currently running pipelines",
)
PIPELINE_DURATION = Histogram(
    "pipeline_duration_seconds",
    "Pipeline execution duration",
    ["pipeline_name"],
)

@app.middleware("http")
async def monitor_requests(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    REQUESTS_TOTAL.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code,
    ).inc()
    REQUESTS_LATENCY.labels(
        method=request.method,
        endpoint=request.url.path,
    ).observe(duration)
    return response

5.2 Grafana Dashboard

{
  "title": "DataPipeline Overview",
  "panels": [
    {
      "title": "Request Rate",
      "type": "graph",
      "targets": [{"expr": "rate(api_requests_total[5m])"}]
    },
    {
      "title": "P95 Latency",
      "type": "graph",
      "targets": [{"expr": "histogram_quantile(0.95, rate(api_request_latency_seconds_bucket[5m]))"}]
    },
    {
      "title": "Active Pipelines",
      "type": "stat",
      "targets": [{"expr": "active_pipelines"}]
    },
    {
      "title": "Pipeline Duration",
      "type": "heatmap",
      "targets": [{"expr": "rate(pipeline_duration_seconds_bucket[1h])"}]
    }
  ]
}

6. Tests

6.1 Tests unitaires

import pytest
from httpx import AsyncClient, ASGITransport
from src.api.main import app

@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

@pytest.mark.asyncio
async def test_health(client):
    response = await client.get("/api/v1/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

6.2 Tests d'integration

import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="module")
def postgres():
    with PostgresContainer("postgres:16-alpine") as pg:
        yield pg

@pytest.fixture(scope="module")
def redis():
    with RedisContainer("redis:7-alpine") as r:
        yield r

@pytest.mark.asyncio
async def test_pipeline_execution(postgres, redis):
    os.environ["DATABASE_URL"] = postgres.get_connection_url()
    os.environ["REDIS_URL"] = redis.get_connection_url()
    # Test pipeline execution

7. CI/CD

7.1 Docker Compose

version: "3.8"
services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: datapipeline
      POSTGRES_USER: app
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes: ["pgdata:/var/lib/postgresql/data"]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app -d datapipeline"]

  redis:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]

  api:
    build: .
    command: uvicorn src.api.main:app --host 0.0.0.0 --port 8000
    ports: ["8000:8000"]
    depends_on: [postgres, redis]
    env_file: .env

  worker:
    build: .
    command: celery -A src.workers.celery_app worker -l info
    depends_on: [postgres, redis]
    env_file: .env

  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.12.0
    command: >
      mlflow server
      --host 0.0.0.0 --port 5000
      --backend-store-uri postgresql://app:${DB_PASSWORD}@postgres/mlflow
      --default-artifact-root ./mlruns
    ports: ["5000:5000"]
    depends_on: [postgres]

  prometheus:
    image: prom/prometheus:latest
    ports: ["9090:9090"]
    volumes: ["./docker/prometheus.yml:/etc/prometheus/prometheus.yml"]

  grafana:
    image: grafana/grafana:latest
    ports: ["3000:3000"]
    depends_on: [prometheus]

volumes:
  pgdata:

8. Documentation

8.1 Structure mkdocs

# mkdocs.yml
site_name: DataPipeline
theme: material
nav:
  - Home: index.md
  - Architecture: architecture.md
  - API Reference: api.md
  - Deployment: deployment.md
  - ADRs: adr/

9. Diagramme de deploiement

Diagramme en cours de génération...

10. Bonnes pratiques

1. Configuration : Pydantic Settings avec .env
2. Async first : FastAPI asynchrone, SQLAlchemy async
3. Idempotence : Les workers peuvent etre relances sans effet de bord
4. Retry : Celery avec backoff exponentiel
5. Monitoring : Tout est instrumente (API, workers, DB)
6. Tests : Unitaires + integration + property-based
7. Documentation : mkdocs + ADR + OpenAPI
8. Securite : JWT, rate limiting, CORS, secrets management