Guide Complet : Sobriété des Données

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

Empreinte = Volume × Durée × (Consommation + Overhead PUE) × Facteur carbone

Exemple :
- 100 To stockés pendant 1 an
- Cloud (région France, 60 gCO2e/kWh)
- PUE moyen de 1.3
- Consommation : 1 W/To

Empreinte = 100 × 8760h × 1W × 1.3 × 60 / 1000000
         = 68.3 kgCO2e/an

Soit ~0.7 kgCO2e/To/an pour du cloud en France
(5-20 kgCO2e/To/an dans des régions plus carbonées)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

Donnée originale : 1 To

Après infrastructure standard :
├─ Copie locale (RAID) : ×1.5 à ×3
├─ Backup quotidien (7 jours) : ×7
├─ Backup mensuel (12 mois) : ×12
├─ Réplication multi-région : ×2-3
└─ Snapshots et versions : ×1.5-3

Volume réel : 10 à 50 To pour 1 To "utile"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

Calcul du coût du dark data :

Hypothèses entreprise moyenne :
- Stockage total : 500 To
- Dark data : 60% = 300 To
- Coût stockage cloud : 20€/To/mois

Coût annuel du dark data :
300 To × 20€ × 12 mois = 72 000€/an

Empreinte carbone (cloud France) :
300 To × 0.7 kgCO2e = 210 kgCO2e/an

Pour une grande entreprise (50 Po de données) :
30 Po de dark data × 20€ × 12 = 7.2 M€/an

1

Collecter tout → Stocker indéfiniment → Peut-être utiliser un jour

1

Définir le besoin → Collecter le minimum → Durée de vie explicite → Suppression automatique

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

❌ Formulaire maximaliste :
- Nom, Prénom, Email, Téléphone, Adresse complète
- Date de naissance, Sexe, Situation familiale
- Profession, Entreprise, Revenus
- "Comment nous avez-vous connu ?"
- Centres d'intérêt (checkboxes multiples)

✅ Formulaire sobre :
- Email (obligatoire, pour contact)
- Prénom (optionnel, pour personnalisation)
- Message (essentiel)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

❌ Tracking maximaliste :
- User ID permanent
- Toutes les pages vues
- Temps passé par pixel
- Mouvements de souris
- Device fingerprint complet

✅ Analytics sobre :
- Pages vues agrégées (pas de user ID)
- Métriques essentielles uniquement
- Agrégation immédiate (pas de données brutes)
- Anonymisation ou pseudonymisation

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

❌ Logging excessif :
log.debug(f"User {user_id} requested {full_request_body}")
log.debug(f"Response: {full_response_body}")
log.debug(f"Processing time for each step: {detailed_timings}")

✅ Logging sobre :
log.info(f"Request: {request_id} {method} {path} -> {status_code} {duration_ms}ms")
# Détails uniquement si erreur
if status_code >= 400:
    log.error(f"Error details: {error_type} - {error_message}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# Redis - TTL automatique
SET session:user123 "data" EX 86400  # Expire en 24h

# Elasticsearch - Index lifecycle
PUT _ilm/policy/logs_policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_age": "7d",
            "max_size": "50gb"
          }
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "shrink": { "number_of_shards": 1 }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

-- PostgreSQL - Partitionnement avec rétention
CREATE TABLE events (
    id BIGSERIAL,
    created_at TIMESTAMP NOT NULL,
    data JSONB
) PARTITION BY RANGE (created_at);

-- Créer des partitions mensuelles
CREATE TABLE events_2024_01 PARTITION OF events
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');

-- Script de purge des anciennes partitions
DO $$
DECLARE
    partition_name TEXT;
    cutoff_date DATE := CURRENT_DATE - INTERVAL '90 days';
BEGIN
    FOR partition_name IN
        SELECT tablename FROM pg_tables
        WHERE tablename LIKE 'events_%'
        AND substring(tablename from 'events_(\d{4}_\d{2})')::DATE < cutoff_date
    LOOP
        EXECUTE format('DROP TABLE %I', partition_name);
        RAISE NOTICE 'Dropped partition: %', partition_name;
    END LOOP;
END $$;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# Job de purge quotidien
import logging
from datetime import datetime, timedelta
from apscheduler.schedulers.blocking import BlockingScheduler

logger = logging.getLogger(__name__)

def purge_old_sessions():
    """Supprime les sessions de plus de 7 jours"""
    cutoff = datetime.utcnow() - timedelta(days=7)
    deleted = db.sessions.delete_many({"last_activity": {"$lt": cutoff}})
    logger.info(f"Purged {deleted.deleted_count} old sessions")
    metrics.record("sessions_purged", deleted.deleted_count)

def purge_old_logs():
    """Archive et supprime les logs de plus de 90 jours"""
    cutoff = datetime.utcnow() - timedelta(days=90)

    # Archiver d'abord
    logs_to_archive = db.logs.find({"timestamp": {"$lt": cutoff}})
    archive_to_cold_storage(logs_to_archive)

    # Puis supprimer
    deleted = db.logs.delete_many({"timestamp": {"$lt": cutoff}})
    logger.info(f"Purged {deleted.deleted_count} old logs")

def purge_orphaned_files():
    """Supprime les fichiers non référencés"""
    referenced_ids = set(db.documents.distinct("file_id"))
    all_files = storage.list_files("uploads/")

    orphans = [f for f in all_files if f.id not in referenced_ids]
    for orphan in orphans:
        storage.delete(orphan)

    logger.info(f"Purged {len(orphans)} orphaned files")

scheduler = BlockingScheduler()
scheduler.add_job(purge_old_sessions, 'cron', hour=2)
scheduler.add_job(purge_old_logs, 'cron', hour=3)
scheduler.add_job(purge_orphaned_files, 'cron', day_of_week='sun', hour=4)
scheduler.start()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

┌─────────────────────────────────────────────────────────────┐
│           Mon espace de stockage - 45 Go utilisés           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Répartition :                                              │
│  ████████████████████░░░░░░░░░░ Documents : 23 Go          │
│  ██████████░░░░░░░░░░░░░░░░░░░░ Emails : 12 Go             │
│  ██████░░░░░░░░░░░░░░░░░░░░░░░░ Photos : 7 Go              │
│  ███░░░░░░░░░░░░░░░░░░░░░░░░░░░ Autres : 3 Go              │
│                                                             │
│  ⚠️ Suggestions de nettoyage :                              │
│  ┌─────────────────────────────────────────────────────────┐│
│  │ 📁 Fichiers volumineux non ouverts depuis 1 an          ││
│  │    12 fichiers - 8.5 Go                    [Voir]       ││
│  ├─────────────────────────────────────────────────────────┤│
│  │ 📧 Emails avec pièces jointes > 5 Mo                    ││
│  │    234 emails - 4.2 Go                     [Voir]       ││
│  ├─────────────────────────────────────────────────────────┤│
│  │ 📋 Doublons détectés                                    ││
│  │    45 fichiers - 2.1 Go                    [Voir]       ││
│  └─────────────────────────────────────────────────────────┘│
│                                                             │
│  Économie potentielle : 14.8 Go                             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1
2
3
4
5

Politique de nommage :
├─ Un seul emplacement de référence par document
├─ Liens plutôt que copies
├─ Versioning automatique (pas de _v2, _final)
└─ Nettoyage régulier des espaces personnels

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Déduplication par hash au stockage
import hashlib

def store_file(content, filename):
    # Calculer le hash du contenu
    file_hash = hashlib.sha256(content).hexdigest()

    # Vérifier si le contenu existe déjà
    existing = storage.get_by_hash(file_hash)
    if existing:
        # Créer seulement une référence
        return create_reference(filename, existing.id)

    # Sinon, stocker le nouveau contenu
    return storage.store(content, file_hash, filename)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

Inventaire technique :
├─ Bases de données
│   ├─ PostgreSQL production : 2.3 To
│   ├─ MongoDB analytics : 5.1 To
│   └─ Redis cache : 50 Go
├─ Stockage fichiers
│   ├─ S3 documents : 12 To
│   ├─ NAS interne : 8 To
│   └─ OneDrive utilisateurs : 15 To
├─ Emails
│   └─ Exchange Online : 4 To
├─ Applications SaaS
│   ├─ Salesforce : 500 Go
│   └─ Slack : 200 Go
└─ Backups
    ├─ Veeam : 45 To
    └─ AWS S3 Glacier : 30 To

TOTAL : ~120 To

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# Schéma de classification
categories:
  - name: "Données personnelles"
    sensitivity: high
    retention: "Durée relation + 3 ans"
    examples:
      - "Données clients CRM"
      - "Données employés RH"

  - name: "Données business"
    sensitivity: medium
    retention: "10 ans (légal)"
    examples:
      - "Factures et contrats"
      - "Transactions"

  - name: "Données techniques"
    sensitivity: low
    retention: "30-90 jours"
    examples:
      - "Logs applicatifs"
      - "Métriques"

  - name: "Données projet"
    sensitivity: medium
    retention: "Durée projet + 3 ans"
    examples:
      - "Documents de travail"
      - "Livrables"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- Analyse d'accès aux fichiers (exemple)
SELECT
    date_trunc('month', last_accessed) as month,
    count(*) as file_count,
    sum(size_bytes) / 1024 / 1024 / 1024 as size_gb
FROM files
GROUP BY 1
ORDER BY 1;

-- Résultat exemple :
-- month      | file_count | size_gb
-- -----------+------------+---------
-- 2024-12    | 15,234     | 45.2    (accès récent)
-- 2024-06    | 8,432      | 23.1    (6 mois)
-- 2023-12    | 12,876     | 67.3    (>1 an = dark data potentiel)
-- Jamais     | 34,521     | 89.4    (dark data certain)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

# Politique de rétention des données

## 1. Objet et périmètre
Cette politique définit les durées de conservation des données
de [Entreprise] et les procédures de suppression associées.

## 2. Principes généraux
- Toute donnée collectée doit avoir une durée de vie définie
- La suppression est automatique sauf exception documentée
- Les dérogations requièrent une validation Data Owner + DPO

## 3. Durées de rétention par catégorie

### 3.1 Données clients
| Sous-catégorie | Durée | Base légale |
|----------------|-------|-------------|
| Données de contact | Relation + 3 ans | Prescription commerciale |
| Historique achats | 10 ans | Code commerce |
| Données de navigation | 13 mois | CNIL |

### 3.2 Données employés
| Sous-catégorie | Durée | Base légale |
|----------------|-------|-------------|
| Dossier personnel | Départ + 5 ans | Prescription sociale |
| Bulletins de paie | Départ + 5 ans | Code du travail |
| Documents médicaux | Départ + 40 ans | Code de la santé |

### 3.3 Données techniques
| Sous-catégorie | Durée | Justification |
|----------------|-------|---------------|
| Logs applicatifs | 90 jours | Debug et investigation |
| Logs sécurité | 2 ans | Conformité ISO 27001 |
| Métriques | 2 ans agrégées, 30j détaillées | Capacity planning |

## 4. Procédures de suppression

### 4.1 Suppression automatique
- Jobs de purge quotidiens pour les données techniques
- Archivage puis suppression pour les données business
- Vérification et audit mensuel des processus

### 4.2 Suppression sur demande
- Traitement des demandes RGPD sous 30 jours
- Validation par le Data Owner avant exécution
- Certificat de suppression fourni

## 5. Exceptions et dérogations
Les dérogations doivent être :
- Documentées avec justification
- Limitées dans le temps
- Validées par Data Owner et DPO
- Révisées annuellement

## 6. Responsabilités
- **Data Owners** : définition des durées métier
- **IT** : implémentation technique des purges
- **DPO** : validation conformité RGPD
- **Audit** : contrôle annuel

## 7. Révision
Cette politique est révisée annuellement ou lors de tout
changement réglementaire significatif.

1
2
3
4
5
6

Rapport automatique mensuel :
├─ Volume total et évolution (+/-%)
├─ Alertes de dépassement de quotas
├─ Purges effectuées (succès/échecs)
├─ Données approchant fin de rétention
└─ Anomalies détectées

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

Pour chaque champ collecté, répondre à :

1. Cette donnée est-elle NÉCESSAIRE pour la finalité déclarée ?
   (pas "utile" ou "pratique" - NÉCESSAIRE)

2. Existe-t-il un moyen d'atteindre l'objectif avec MOINS de données ?
   (agrégation, pseudonymisation, échantillonnage)

3. La granularité est-elle JUSTIFIÉE ?
   (date de naissance complète vs tranche d'âge)

Si la réponse est "non" à 1 ou "oui" à 2-3 → ne pas collecter

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

Réception demande (J)
    ↓
Vérification identité (J à J+3)
    ↓
Analyse du périmètre (J+3 à J+10)
├─ Quelles données ?
├─ Quels systèmes ?
├─ Exceptions applicables ?
    ↓
Exécution de l'effacement (J+10 à J+25)
├─ Bases de production
├─ Backups (si techniquement possible)
├─ Archives
├─ Sous-traitants
    ↓
Confirmation à la personne (J+30 max)
├─ Liste des données effacées
├─ Données conservées (avec justification)

1
2

Avant : Jean Dupont, 35 ans, Paris 15e
Après : H, 30-40 ans, Paris

1
2

Avant : Achats individuels par client
Après : Moyenne des achats par tranche d'âge et région

1
2
3
4
5
6

# K-anonymité par ajout de bruit
import numpy as np

def add_noise(value, epsilon=0.1):
    noise = np.random.laplace(0, 1/epsilon)
    return value + noise

1
2

Avant : {nom: "Dupont", email: "j.dupont@mail.com", age: 35}
Après : {age: 35}  # Identifiants directs supprimés

1
2
3
4
5
6
7
8
9

# Test simple de k-anonymité
def check_k_anonymity(df, quasi_identifiers, k=5):
    """Vérifie que chaque combinaison apparaît au moins k fois"""
    counts = df.groupby(quasi_identifiers).size()
    return counts.min() >= k

# Usage
quasi_ids = ['age_group', 'region', 'gender']
is_k_anonymous = check_k_anonymity(anonymized_df, quasi_ids, k=5)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# logging.yaml
production:
  level: INFO
  filters:
    - exclude_debug_modules
  handlers:
    - rotating_file:
        max_size: 100MB
        backup_count: 30
    - elasticsearch:
        index_pattern: "logs-prod-{date}"
        retention_days: 90

development:
  level: DEBUG
  handlers:
    - console
    - file:
        path: /tmp/app.log

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# ❌ Log non structuré
logging.info(f"User {user_id} logged in from {ip} at {timestamp}")

# ✅ Log structuré
logging.info("user_login", extra={
    "user_id": user_id,
    "ip": ip,
    "timestamp": timestamp,
    "event_type": "authentication"
})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# Échantillonnage pour logs à haut volume
import random

class SampledLogger:
    def __init__(self, logger, sample_rate=0.1):
        self.logger = logger
        self.sample_rate = sample_rate

    def info(self, msg, **kwargs):
        if random.random() < self.sample_rate:
            self.logger.info(msg, **kwargs)

# Agrégation en mémoire avant écriture
from collections import defaultdict
import threading

class AggregatingLogger:
    def __init__(self, flush_interval=60):
        self.counts = defaultdict(int)
        self.lock = threading.Lock()

    def count(self, event_type):
        with self.lock:
            self.counts[event_type] += 1

    def flush(self):
        with self.lock:
            for event_type, count in self.counts.items():
                logging.info(f"{event_type}: {count} occurrences")
            self.counts.clear()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# ❌ Session volumineuse
session['user'] = {
    'id': 123,
    'profile': {...},  # Données complètes
    'orders': [...],   # Historique complet
    'preferences': {...},
    'cart': [...],
    'viewed_products': [...],  # Historique navigation
}

# ✅ Session sobre
session['user_id'] = 123
session['roles'] = ['customer']
session['cart_id'] = 'cart_abc123'  # Référence, pas les données

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# Configuration Flask avec expiration
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=2)
app.config['SESSION_REFRESH_EACH_REQUEST'] = True

# Redis avec TTL automatique
@app.before_request
def refresh_session():
    session.modified = True  # Reset le TTL

# Nettoyage des sessions orphelines
def cleanup_orphaned_sessions():
    """Supprime les sessions sans activité depuis 24h"""
    cutoff = datetime.utcnow() - timedelta(hours=24)
    redis.delete(*redis.keys('session:*'))  # À adapter

1
2
3
4
5
6

# redis.conf
maxmemory 2gb
maxmemory-policy allkeys-lru  # Supprime les moins récemment utilisés

# Ou volatile-ttl pour ne supprimer que les clés avec TTL
maxmemory-policy volatile-ttl

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# Nettoyage des fichiers temporaires
import os
import time
from pathlib import Path

def cleanup_temp_files(temp_dir, max_age_hours=24):
    """Supprime les fichiers temporaires de plus de X heures"""
    cutoff = time.time() - (max_age_hours * 3600)

    for file_path in Path(temp_dir).rglob('*'):
        if file_path.is_file():
            if file_path.stat().st_mtime < cutoff:
                file_path.unlink()
                logging.info(f"Deleted temp file: {file_path}")

# Job Kubernetes pour nettoyage régulier
# cronjob.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
  name: cleanup-temp-files
spec:
  schedule: "0 */6 * * *"  # Toutes les 6 heures
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: cleanup
            image: busybox
            command:
            - find
            - /tmp
            - -type
            - f
            - -mtime
            - "+1"
            - -delete
          restartPolicy: OnFailure

1
2
3
4
5
6
7
8

Environnements strictement séparés :
├─ Production : données réelles, jamais de test
├─ Staging : copie anonymisée ou données synthétiques
├─ Test : données synthétiques ou fixtures
└─ Développement : fixtures minimales

❌ JAMAIS de données de production en dev/test
❌ JAMAIS de données de test en production

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# Génération de données réalistes mais fausses
from faker import Faker

fake = Faker('fr_FR')

def generate_test_users(n=100):
    return [
        {
            'id': i,
            'email': fake.email(),
            'name': fake.name(),
            'address': fake.address(),
            'phone': fake.phone_number(),
            'created_at': fake.date_time_this_year()
        }
        for i in range(n)
    ]

def generate_test_orders(users, n=500):
    return [
        {
            'id': i,
            'user_id': fake.random_element(users)['id'],
            'amount': fake.pydecimal(min_value=10, max_value=500, right_digits=2),
            'status': fake.random_element(['pending', 'paid', 'shipped']),
            'created_at': fake.date_time_this_year()
        }
        for i in range(n)
    ]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# Fixture avec nettoyage automatique (pytest)
import pytest

@pytest.fixture
def test_data(db):
    """Crée des données de test et les nettoie après"""
    # Setup
    users = create_test_users(10)
    orders = create_test_orders(users, 50)

    yield {'users': users, 'orders': orders}

    # Teardown automatique
    db.orders.delete_many({'id': {'$in': [o['id'] for o in orders]}})
    db.users.delete_many({'id': {'$in': [u['id'] for u in users]}})

# CI/CD : nettoyage de l'environnement de test
# .gitlab-ci.yml
cleanup_test_env:
  stage: cleanup
  script:
    - python scripts/cleanup_test_data.py
  only:
    - schedules  # Nettoyage quotidien planifié

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

-- Exemple : stocker le nom du client dans la commande
-- Évite une jointure fréquente, mais duplique l'info

-- ❌ Dénormalisation excessive
CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer_id INT,
    customer_name VARCHAR(100),      -- Dupliqué
    customer_email VARCHAR(255),     -- Dupliqué
    customer_phone VARCHAR(20),      -- Dupliqué
    customer_address TEXT,           -- Dupliqué
    -- ... et le reste de la commande
);

-- ✅ Dénormalisation raisonnable
CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer_id INT REFERENCES customers(id),
    customer_name_snapshot VARCHAR(100), -- Seulement ce qui est utile et stable
    -- ... reste de la commande
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

-- Exemple d'optimisation
-- ❌ Avant
CREATE TABLE events (
    id VARCHAR(36),           -- UUID en texte
    event_type VARCHAR(50),   -- Texte libre
    is_processed VARCHAR(5),  -- "true"/"false"
    created_at VARCHAR(30)    -- "2024-01-15T10:30:00Z"
);

-- ✅ Après
CREATE TABLE events (
    id UUID DEFAULT gen_random_uuid(),
    event_type SMALLINT,      -- Référence vers table de types
    is_processed BOOLEAN DEFAULT false,
    created_at TIMESTAMPTZ DEFAULT now()
);

-- Économie : ~70% d'espace par ligne

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

-- PostgreSQL : table partitionnée par mois
CREATE TABLE events (
    id BIGSERIAL,
    event_type VARCHAR(50),
    payload JSONB,
    created_at TIMESTAMPTZ NOT NULL
) PARTITION BY RANGE (created_at);

-- Partitions par mois
CREATE TABLE events_2024_01 PARTITION OF events
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE events_2024_02 PARTITION OF events
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');
-- etc.

-- Création automatique des partitions futures (pg_partman)
SELECT partman.create_parent(
    p_parent_table := 'public.events',
    p_control := 'created_at',
    p_type := 'native',
    p_interval := '1 month',
    p_premake := 3  -- Créer 3 mois à l'avance
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# Script d'archivage des anciennes partitions
import subprocess
from datetime import datetime, timedelta

def archive_old_partition(table_name, partition_date):
    """Archive une partition vers S3 puis la supprime"""

    partition_name = f"{table_name}_{partition_date.strftime('%Y_%m')}"

    # Export vers fichier compressé
    export_cmd = f"""
        pg_dump -t {partition_name} --data-only --format=custom
        | gzip > /tmp/{partition_name}.dump.gz
    """
    subprocess.run(export_cmd, shell=True, check=True)

    # Upload vers S3 Glacier
    subprocess.run([
        'aws', 's3', 'cp',
        f'/tmp/{partition_name}.dump.gz',
        f's3://archives/database/{partition_name}.dump.gz',
        '--storage-class', 'DEEP_ARCHIVE'
    ], check=True)

    # Supprimer la partition
    subprocess.run([
        'psql', '-c', f'DROP TABLE {partition_name}'
    ], check=True)

# Archiver les partitions de plus de 1 an
cutoff = datetime.now() - timedelta(days=365)
archive_old_partition('events', cutoff)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

# Job de purge avec monitoring
import logging
from datetime import datetime, timedelta
from contextlib import contextmanager

logger = logging.getLogger(__name__)

class DataPurger:
    def __init__(self, db):
        self.db = db
        self.stats = {'deleted': 0, 'errors': 0}

    @contextmanager
    def track_purge(self, table_name):
        start = datetime.now()
        try:
            yield
            duration = (datetime.now() - start).total_seconds()
            logger.info(f"Purge {table_name}: {self.stats['deleted']} rows in {duration:.2f}s")
        except Exception as e:
            self.stats['errors'] += 1
            logger.error(f"Purge {table_name} failed: {e}")
            raise

    def purge_old_sessions(self, days=7):
        """Supprime les sessions de plus de X jours"""
        with self.track_purge('sessions'):
            cutoff = datetime.utcnow() - timedelta(days=days)
            result = self.db.execute(
                "DELETE FROM sessions WHERE last_activity < %s",
                [cutoff]
            )
            self.stats['deleted'] = result.rowcount

    def purge_old_logs(self, days=90):
        """Supprime les logs de plus de X jours par batch"""
        with self.track_purge('logs'):
            cutoff = datetime.utcnow() - timedelta(days=days)
            total_deleted = 0

            while True:
                # Supprimer par batch de 10000 pour éviter les locks longs
                result = self.db.execute("""
                    DELETE FROM logs
                    WHERE id IN (
                        SELECT id FROM logs
                        WHERE created_at < %s
                        LIMIT 10000
                    )
                """, [cutoff])

                total_deleted += result.rowcount
                if result.rowcount < 10000:
                    break

            self.stats['deleted'] = total_deleted

    def purge_soft_deleted(self, days=30):
        """Supprime définitivement les enregistrements soft-deleted"""
        with self.track_purge('soft_deleted'):
            cutoff = datetime.utcnow() - timedelta(days=days)

            for table in ['users', 'orders', 'documents']:
                result = self.db.execute(f"""
                    DELETE FROM {table}
                    WHERE deleted_at IS NOT NULL
                    AND deleted_at < %s
                """, [cutoff])
                self.stats['deleted'] += result.rowcount

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# Alerting sur les purges (Prometheus)
groups:
  - name: data_purge_alerts
    rules:
      - alert: PurgeJobFailed
        expr: purge_job_last_success_timestamp < (time() - 86400)
        for: 1h
        labels:
          severity: warning
        annotations:
          summary: "Purge job hasn't run successfully in 24h"

      - alert: DataRetentionExceeded
        expr: data_age_days > data_retention_policy_days
        for: 1h
        labels:
          severity: critical
        annotations:
          summary: "Data older than retention policy exists"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- Analyser l'utilisation des index
SELECT
    schemaname,
    tablename,
    indexname,
    idx_scan as times_used,
    idx_tup_read,
    idx_tup_fetch,
    pg_size_pretty(pg_relation_size(indexrelid)) as size
FROM pg_stat_user_indexes
ORDER BY idx_scan ASC;

-- Index jamais utilisés = candidats à la suppression
-- Attention : vérifier sur une période représentative (semaines/mois)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- ❌ Index sur colonnes peu sélectives
CREATE INDEX idx_status ON orders(status);  -- Si 90% sont "completed"

-- ❌ Index dupliqués
CREATE INDEX idx_user ON orders(user_id);
CREATE INDEX idx_user_date ON orders(user_id, created_at);
-- Le second couvre déjà les requêtes du premier

-- ❌ Index sur petites tables
CREATE INDEX idx_countries ON countries(code);  -- Table de 200 lignes

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- ✅ Index composites pour les requêtes fréquentes
CREATE INDEX idx_orders_user_date ON orders(user_id, created_at DESC);

-- ✅ Index partiels pour les données actives
CREATE INDEX idx_orders_pending ON orders(created_at)
WHERE status = 'pending';

-- ✅ Index couvrants pour éviter les accès table
CREATE INDEX idx_orders_summary ON orders(user_id)
INCLUDE (total_amount, status);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

storage/
├── documents/
│   ├── contracts/
│   │   ├── 2024/
│   │   │   ├── 01/
│   │   │   └── 02/
│   │   └── 2023/
│   └── invoices/
│       └── {year}/{month}/
├── media/
│   ├── images/
│   │   ├── originals/   # Conservés temporairement
│   │   ├── processed/   # Versions optimisées
│   │   └── thumbnails/  # Vignettes
│   └── videos/
├── temp/                 # Purge automatique < 24h
│   └── uploads/
└── archives/             # Stockage froid
    └── {year}/

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

Format recommandé : {type}_{id}_{version}_{date}.{ext}

Exemples :
- invoice_12345_v1_20240115.pdf
- contract_abc123_signed_20240110.pdf
- profile_user789_thumb_100x100.jpg

Avantages :
- Tri chronologique naturel
- Identification sans ouvrir
- Versioning explicite
- Unicité garantie

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

import hashlib
from pathlib import Path

class DeduplicatedStorage:
    def __init__(self, base_path):
        self.base_path = Path(base_path)
        self.content_path = self.base_path / 'content'
        self.refs_path = self.base_path / 'refs'

    def store(self, file_content, logical_path):
        """Stocke un fichier avec déduplication"""
        # Calculer le hash du contenu
        content_hash = hashlib.sha256(file_content).hexdigest()
        content_file = self.content_path / content_hash[:2] / content_hash

        # Stocker le contenu s'il n'existe pas
        if not content_file.exists():
            content_file.parent.mkdir(parents=True, exist_ok=True)
            content_file.write_bytes(file_content)

        # Créer une référence vers le contenu
        ref_file = self.refs_path / logical_path
        ref_file.parent.mkdir(parents=True, exist_ok=True)
        ref_file.write_text(content_hash)

        return content_hash

    def get(self, logical_path):
        """Récupère un fichier par son chemin logique"""
        ref_file = self.refs_path / logical_path
        if not ref_file.exists():
            raise FileNotFoundError(logical_path)

        content_hash = ref_file.read_text()
        content_file = self.content_path / content_hash[:2] / content_hash
        return content_file.read_bytes()

    def delete(self, logical_path):
        """Supprime une référence (pas le contenu si encore utilisé)"""
        ref_file = self.refs_path / logical_path
        if ref_file.exists():
            ref_file.unlink()

    def cleanup_orphans(self):
        """Supprime les contenus sans référence"""
        # Collecter tous les hashs référencés
        referenced = set()
        for ref_file in self.refs_path.rglob('*'):
            if ref_file.is_file():
                referenced.add(ref_file.read_text())

        # Supprimer les contenus non référencés
        deleted = 0
        for content_file in self.content_path.rglob('*'):
            if content_file.is_file():
                content_hash = content_file.name
                if content_hash not in referenced:
                    content_file.unlink()
                    deleted += 1

        return deleted

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from PIL import Image
from pathlib import Path

def optimize_image(input_path, output_path, max_size=(1920, 1080), quality=85):
    """Optimise une image : redimensionne et compresse"""
    img = Image.open(input_path)

    # Redimensionner si nécessaire
    if img.width > max_size[0] or img.height > max_size[1]:
        img.thumbnail(max_size, Image.LANCZOS)

    # Convertir en RGB si nécessaire (pour JPEG)
    if img.mode in ('RGBA', 'P'):
        img = img.convert('RGB')

    # Sauvegarder avec compression
    img.save(output_path, 'JPEG', quality=quality, optimize=True)

    # Stats
    original_size = Path(input_path).stat().st_size
    new_size = Path(output_path).stat().st_size
    reduction = (1 - new_size / original_size) * 100

    return {'original': original_size, 'new': new_size, 'reduction': f'{reduction:.1f}%'}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import boto3
from datetime import datetime, timedelta

s3 = boto3.client('s3')

# Politique de lifecycle
lifecycle_config = {
    'Rules': [
        {
            'ID': 'ArchiveOldFiles',
            'Status': 'Enabled',
            'Filter': {'Prefix': 'documents/'},
            'Transitions': [
                {
                    'Days': 30,
                    'StorageClass': 'STANDARD_IA'  # Infrequent Access
                },
                {
                    'Days': 90,
                    'StorageClass': 'GLACIER_IR'   # Glacier Instant Retrieval
                },
                {
                    'Days': 365,
                    'StorageClass': 'DEEP_ARCHIVE' # Archive long terme
                }
            ],
            'Expiration': {
                'Days': 2555  # Suppression après 7 ans
            }
        }
    ]
}

s3.put_bucket_lifecycle_configuration(
    Bucket='my-bucket',
    LifecycleConfiguration=lifecycle_config
)

1
2
3
4
5
6
7
8

Acquisition  : D'où viennent les utilisateurs ?
Activation   : Ont-ils une bonne première expérience ?
Retention    : Reviennent-ils ?
Referral     : Recommandent-ils ?
Revenue      : Génèrent-ils du revenu ?

Pour chaque étape : 2-3 métriques clés maximum
Pas de "nice to have" qui ne sera jamais analysé

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

❌ Collecte excessive :
- Toutes les pages vues avec détail
- Tous les clics avec coordonnées
- Temps passé par élément
- Mouvements de souris
- Historique complet par utilisateur

✅ Collecte sobre :
- Pages vues (agrégées par page)
- Conversions (événements clés)
- Origine du trafic (agrégée)
- Temps moyen par session

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

# Pipeline d'agrégation progressive
class AnalyticsPipeline:
    def __init__(self):
        self.raw_retention = timedelta(days=7)
        self.hourly_retention = timedelta(days=30)
        self.daily_retention = timedelta(days=365)
        self.monthly_retention = timedelta(days=365*5)

    def ingest_event(self, event):
        """Ingère un événement brut"""
        # Stockage temporaire des événements bruts
        self.raw_store.store(event)

        # Mise à jour des agrégats en temps réel
        self.update_realtime_counters(event)

    def hourly_aggregation(self):
        """Job horaire : agrège les données brutes"""
        hour = datetime.utcnow().replace(minute=0, second=0, microsecond=0)
        events = self.raw_store.get_events(hour - timedelta(hours=1), hour)

        aggregates = {
            'timestamp': hour,
            'page_views': defaultdict(int),
            'unique_visitors': set(),
            'conversions': defaultdict(int),
        }

        for event in events:
            if event['type'] == 'page_view':
                aggregates['page_views'][event['page']] += 1
            aggregates['unique_visitors'].add(event['session_id'])
            if event.get('conversion'):
                aggregates['conversions'][event['conversion_type']] += 1

        # Transformer les sets en counts
        aggregates['unique_visitors'] = len(aggregates['unique_visitors'])

        self.hourly_store.store(aggregates)

    def daily_aggregation(self):
        """Job quotidien : agrège les données horaires"""
        date = datetime.utcnow().date() - timedelta(days=1)
        hourly_data = self.hourly_store.get_day(date)

        daily_aggregate = {
            'date': date,
            'total_page_views': sum(h['page_views'].values() for h in hourly_data),
            'unique_visitors': ...,  # Recalculer depuis les raw ou estimer
            'conversions': ...,
        }

        self.daily_store.store(daily_aggregate)

    def cleanup_old_data(self):
        """Supprime les données au-delà de leur rétention"""
        now = datetime.utcnow()
        self.raw_store.delete_before(now - self.raw_retention)
        self.hourly_store.delete_before(now - self.hourly_retention)
        self.daily_store.delete_before(now - self.daily_retention)