🤖 Data Engineering for AI
Module : Data Engineering for AI
5. Automatiser et Versionner
Durée estimée : 4-5 heures
Prérequis : Module 4 (Enrichir) + Docker installé
📍 Où en es-tu dans le parcours ?
DONNÉES BRUTES DATASET PRÊT
(chaos) (training-ready)
│ ▲
│ Module 2 Module 3 Module 4 Module 5 │
│ ────────── ────────── ────────── ────────── │
│ │
▼ │
┌─────────┐ ┌─────────────┐ ┌──────────┐ ┌────────┴───────┐
│ │ │ │ │ │ │ │
│ STOCKER │─────▶│ TRANSFORMER │─────▶│ ENRICHIR │─────▶│ AUTOMATISER & │
│ ✅ │ │ ✅ │ │ ✅ │ │ VERSIONNER │
│ │ │ │ │ │ │ ▲▲▲▲ │
└─────────┘ └─────────────┘ └──────────┘ └────────────────┘
TU ES ICI
🎯 Question de ce module
Comment rendre le pipeline reproductible, automatique et traçable ?
📦 Ce que tu vas livrer
À la fin de ce module, tu auras :
- Un DAG Airflow qui orchestre tout le pipeline
- Un versioning DVC pour tracker les données
- Un tracking MLflow pour les métadonnées du dataset
- Un pipeline 100% reproductible
🛠️ Ce que tu vas apprendre
| Compétence | Pourquoi c'est important | Outil |
|---|---|---|
| Orchestration | Exécuter les tâches dans le bon ordre, automatiquement | Airflow |
| Versioning données | Revenir à une version précédente du dataset | DVC |
| Tracking | Savoir quel dataset a produit quel modèle | MLflow |
⚠️ Scope du Data Engineer
┌─────────────────────────────────────────────────────────────────────────────┐
│ QUI FAIT QUOI ? │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ DATA ENGINEER ML ENGINEER │
│ ───────────── ─────────── │
│ │
│ ✅ Orchestrer le pipeline DATA ✅ Orchestrer le pipeline TRAINING │
│ ✅ Versionner les DONNÉES ✅ Versionner les MODÈLES │
│ ✅ Tracker les DATASETS ✅ Tracker les EXPÉRIENCES │
│ │
│ Ingestion → Transform → Enrichir Training → Évaluation → Déploiement │
│ │
│ Ce module se concentre sur la partie DATA ENGINEERING │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
5.0 Pourquoi automatiser ?
Le problème : le pipeline manuel
┌─────────────────────────────────────────────────────────────────────────────┐
│ PIPELINE MANUEL vs AUTOMATISÉ │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ MANUEL (Modules 2-3-4) AUTOMATISÉ (Module 5) │
│ ────────────────────── ───────────────────── │
│ │
│ 1. Lancer le script d'ingestion 1. Déclencher le DAG │
│ 2. Attendre... 2. Tout s'exécute automatiquement │
│ 3. Lancer le script de transform 3. Recevoir une notif "Terminé" │
│ 4. Attendre... │
│ 5. Lancer le script d'enrichment Avantages: │
│ 6. Attendre... • Pas d'oubli d'étape │
│ 7. Vérifier les résultats • Retry automatique si erreur │
│ 8. Oublier une étape → 💥 • Historique des exécutions │
│ • Parallélisation possible │
│ "C'est qui qui a lancé quoi ?" • Reproductible │
│ "On en est où ?" │
│ "Ça a crashé, je recommence..." │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
Pourquoi versionner les données ?
┌─────────────────────────────────────────────────────────────────────────────┐
│ POURQUOI VERSIONNER ? │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ Scénario: Ton modèle v2 est MOINS BON que v1. Pourquoi ? │
│ │
│ SANS VERSIONING AVEC VERSIONING (DVC) │
│ ──────────────── ───────────────────── │
│ │
│ "C'était quoi le dataset v1 ?" git checkout v1 │
│ "Aucune idée..." dvc checkout │
│ "On a écrasé les données..." → Dataset v1 restauré ! │
│ │
│ ════════════════════════════════════════════════════════════════════ │
│ │
│ Cas d'usage: │
│ • Revenir à une version précédente │
│ • Comparer deux versions du dataset │
│ • Reproduire un entraînement passé │
│ • Auditer : "quel dataset a produit ce modèle ?" │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
Git vs DVC : pourquoi les deux ?
| Outil | Ce qu'il versionne | Stockage |
|---|---|---|
| Git | Code, configs, petits fichiers | GitHub, GitLab |
| DVC | Données, modèles, gros fichiers | S3, MinIO, GCS |
DVC stocke un pointeur (hash) dans Git, et les vraies données dans un stockage distant.
Pourquoi tracker avec MLflow ?
┌─────────────────────────────────────────────────────────────────────────────┐
│ POURQUOI TRACKER ? │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ Question: "Ce dataset, il contient combien d'images ? Quels traitements ?"│
│ │
│ SANS TRACKING AVEC TRACKING (MLflow) │
│ ────────────── ────────────────────── │
│ │
│ "Regarde dans les logs..." mlflow.log_params({ │
│ "Quel fichier de log ?" 'n_images': 50000, │
│ "C'était en prod ou en dev ?" 'resize_mode': 'PAD', │
│ "Je sais plus..." 'target_size': 256, │
│ 'dedup_threshold': 0.95 │
│ }) │
│ │
│ → UI web avec tous les runs │
│ → Comparaison entre versions │
│ → Traçabilité complète │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
💡 Pour le DE : MLflow track les métadonnées du dataset (taille, paramètres de transformation, stats). Le ML Engineer l'utilise pour tracker les expériences de training.
5.1 Setup de l'environnement
Docker Compose complet
# docker-compose.yml
version: '3.8'
services:
# ========== STOCKAGE ==========
minio:
image: minio/minio:latest
container_name: minio
ports:
- "9000:9000"
- "9001:9001"
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin123
command: server /data --console-address ":9001"
volumes:
- minio_data:/data
# ========== ORCHESTRATION ==========
airflow:
image: apache/airflow:2.7.0-python3.10
container_name: airflow
ports:
- "8080:8080"
environment:
- AIRFLOW__CORE__EXECUTOR=LocalExecutor
- AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=sqlite:////opt/airflow/airflow.db
- AIRFLOW__CORE__LOAD_EXAMPLES=False
volumes:
- ./dags:/opt/airflow/dags
- ./logs:/opt/airflow/logs
command: standalone
# ========== TRACKING ==========
mlflow:
image: ghcr.io/mlflow/mlflow:v2.8.0
container_name: mlflow
ports:
- "5000:5000"
command: mlflow server --host 0.0.0.0 --port 5000
volumes:
- mlflow_data:/mlflow
volumes:
minio_data:
mlflow_data:
# Lancer l'environnement
docker-compose up -d
# Vérifier
# MinIO: http://localhost:9001
# Airflow: http://localhost:8080 (admin/admin)
# MLflow: http://localhost:5000
# Installer les dépendances Python
pip install apache-airflow dvc mlflow boto3
5.2 Versionner les données avec DVC
Comment DVC fonctionne ?
┌─────────────────────────────────────────────────────────────────────────────┐
│ COMMENT DVC FONCTIONNE │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ AVANT DVC AVEC DVC │
│ ───────── ──────── │
│ │
│ Git repo: Git repo: │
│ ├── code.py ├── code.py │
│ └── data/ ❌ Trop gros ├── data.dvc ← Pointeur (1KB) │
│ └── images/ pour Git! └── .dvc/ │
│ └── 50GB └── config │
│ │
│ Remote storage (MinIO/S3): │
│ └── .dvc/cache/ ← Vraies données │
│ └── ab/cd1234... (50GB) │
│ │
│ ════════════════════════════════════════════════════════════════════ │
│ │
│ Workflow: │
│ 1. dvc add data/ → Crée data.dvc (hash des fichiers) │
│ 2. git add data.dvc → Versionne le pointeur │
│ 3. dvc push → Upload les données vers MinIO │
│ 4. git push → Push le code + pointeur │
│ │
│ Pour restaurer: │
│ 1. git checkout v1.0 → Récupère data.dvc de la v1.0 │
│ 2. dvc checkout → Télécharge les données correspondantes │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
pythonVoir le code
# ============================================
# Initialiser DVC
# ============================================
# Ces commandes s'exécutent dans le terminal, pas dans Jupyter
# On les affiche ici pour documentation
dvc_commands = '''
# 1. Initialiser un repo Git (si pas déjà fait)
git init
# 2. Initialiser DVC
dvc init
# 3. Configurer le remote (MinIO)
dvc remote add -d minio s3://dvc-storage
dvc remote modify minio endpointurl http://localhost:9000
dvc remote modify minio access_key_id minioadmin
dvc remote modify minio secret_access_key minioadmin123
# 4. Tracker un dossier de données
dvc add data/processed-images
# 5. Commit le pointeur
git add data/processed-images.dvc .gitignore
git commit -m "Add processed images v1"
# 6. Push les données vers MinIO
dvc push
# 7. Tag la version
git tag -a v1.0 -m "Dataset v1.0"
'''
print("📋 Commandes DVC à exécuter dans le terminal:")
print(dvc_commands)pythonVoir le code
# ============================================
# Utiliser DVC depuis Python
# ============================================
import subprocess
import os
class DVCManager:
"""
Wrapper Python pour les commandes DVC.
Utile pour intégrer DVC dans un pipeline automatisé.
"""
def __init__(self, repo_path: str = "."):
self.repo_path = repo_path
def _run(self, cmd: list) -> str:
"""Exécute une commande DVC."""
result = subprocess.run(
["dvc"] + cmd,
cwd=self.repo_path,
capture_output=True,
text=True
)
if result.returncode != 0:
raise Exception(f"DVC error: {result.stderr}")
return result.stdout
def add(self, path: str) -> str:
"""Track un fichier/dossier avec DVC."""
return self._run(["add", path])
def push(self) -> str:
"""Push les données vers le remote."""
return self._run(["push"])
def pull(self) -> str:
"""Pull les données depuis le remote."""
return self._run(["pull"])
def checkout(self) -> str:
"""Restaure les données selon les fichiers .dvc."""
return self._run(["checkout"])
def status(self) -> str:
"""Affiche le status DVC."""
return self._run(["status"])
print("✅ DVCManager défini")
print("\n💡 Exemple d'utilisation:")
print('''
dvc = DVCManager("/path/to/repo")
dvc.add("data/processed-images")
dvc.push()
''')5.3 Tracker avec MLflow
MLflow pour le Data Engineer
┌─────────────────────────────────────────────────────────────────────────────┐
│ MLFLOW POUR LE DE │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ Ce que le DE track: Ce que le MLE track: │
│ ────────────────── ──────────────────── │
│ │
│ • Nombre d'images • Hyperparamètres │
│ • Paramètres de transformation • Métriques (accuracy, loss) │
│ • Taux de rejet • Courbes d'apprentissage │
│ • Nombre de duplicatas • Modèle final │
│ • Version du dataset • Artifacts (checkpoints) │
│ • Durée du pipeline │
│ │
│ ════════════════════════════════════════════════════════════════════ │
│ │
│ Concepts MLflow: │
│ │
│ • Experiment: Groupe de runs (ex: "data-pipeline-prod") │
│ • Run: Une exécution du pipeline │
│ • Parameters: Configs (target_size=256) │
│ • Metrics: Valeurs numériques (n_images=50000) │
│ • Artifacts: Fichiers (catalog.parquet) │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
pythonVoir le code
# ============================================
# Configurer MLflow
# ============================================
import mlflow
from datetime import datetime
# Configurer le tracking server
MLFLOW_TRACKING_URI = "http://localhost:5000"
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
# Créer ou récupérer l'expérience
EXPERIMENT_NAME = "data-engineering-pipeline"
mlflow.set_experiment(EXPERIMENT_NAME)
print(f"✅ MLflow configuré")
print(f" Tracking URI: {MLFLOW_TRACKING_URI}")
print(f" Experiment: {EXPERIMENT_NAME}")pythonVoir le code
# ============================================
# DatasetTracker : wrapper MLflow pour le DE
# ============================================
class DatasetTracker:
"""
Tracker MLflow spécialisé pour le Data Engineering.
Track les métadonnées du dataset, pas les expériences ML.
"""
def __init__(self, run_name: str = None):
self.run_name = run_name or f"pipeline-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
self.run = None
def __enter__(self):
"""Démarre un run MLflow."""
self.run = mlflow.start_run(run_name=self.run_name)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Termine le run."""
mlflow.end_run()
def log_pipeline_config(self, config: dict):
"""Log la configuration du pipeline."""
mlflow.log_params(config)
print(f"📝 Config loggée: {config}")
def log_dataset_stats(self, stats: dict):
"""Log les statistiques du dataset."""
mlflow.log_metrics(stats)
print(f"📊 Stats loggées: {stats}")
def log_artifact(self, file_path: str):
"""Log un fichier (catalog, rapport, etc.)."""
mlflow.log_artifact(file_path)
print(f"📁 Artifact loggé: {file_path}")
def log_step_duration(self, step_name: str, duration_seconds: float):
"""Log la durée d'une étape."""
mlflow.log_metric(f"duration_{step_name}_seconds", duration_seconds)
print("✅ DatasetTracker défini")pythonVoir le code
# ============================================
# Exemple d'utilisation du tracker
# ============================================
# Simuler un run de pipeline
with DatasetTracker(run_name="demo-pipeline") as tracker:
# 1. Log la configuration
tracker.log_pipeline_config({
'source_bucket': 'raw-images',
'target_size': 256,
'resize_mode': 'PAD',
'dedup_threshold': 0.95,
'normalize_range': '[-1, 1]'
})
# 2. Simuler le pipeline et log les stats
tracker.log_dataset_stats({
'n_images_input': 1000,
'n_images_output': 950,
'n_rejected': 30,
'n_duplicates': 20,
'success_rate': 95.0
})
# 3. Log les durées
tracker.log_step_duration('ingestion', 120.5)
tracker.log_step_duration('transformation', 340.2)
tracker.log_step_duration('enrichment', 560.8)
print("\n✅ Run terminé ! Voir http://localhost:5000")5.4 Orchestrer avec Airflow
C'est quoi un DAG ?
┌─────────────────────────────────────────────────────────────────────────────┐
│ DAG = DIRECTED ACYCLIC GRAPH │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ Un DAG définit: │
│ • Les TÂCHES à exécuter │
│ • L'ORDRE d'exécution (dépendances) │
│ • La FRÉQUENCE (schedule) │
│ │
│ Exemple de notre pipeline: │
│ │
│ ┌─────────────┐ │
│ │ Ingest │ │
│ └──────┬──────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Transform │ │
│ └──────┬──────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Dedup │ │ Enrich │ │ Catalog │ ← Parallèle ! │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ └────────────┼────────────┘ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Report │ │
│ └─────────────┘ │
│ │
│ Acyclic = pas de boucle (A → B → A interdit) │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
pythonVoir le code
# ============================================
# DAG Airflow pour le pipeline Data Engineering
# ============================================
# Ce fichier doit être placé dans le dossier dags/ d'Airflow
dag_code = '''
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
# Configuration par défaut
default_args = {
'owner': 'data-engineering',
'depends_on_past': False,
'start_date': datetime(2024, 1, 1),
'retries': 2,
'retry_delay': timedelta(minutes=5),
}
# Définir le DAG
dag = DAG(
dag_id='data_engineering_pipeline',
default_args=default_args,
description='Pipeline complet DE for AI',
schedule_interval='@daily', # Tous les jours à minuit
catchup=False,
)
# ============================================
# TÂCHES
# ============================================
def ingest_data(**context):
"""Étape 1: Ingestion des données brutes."""
print("📥 Ingestion des données...")
# Ici: appeler le code du Module 2
# from src.ingestion import run_ingestion
# run_ingestion()
return {"n_images": 1000}
def transform_data(**context):
"""Étape 2: Transformation."""
print("🔄 Transformation des données...")
# Ici: appeler le code du Module 3
# from src.transform import run_transform
# run_transform()
return {"n_transformed": 950}
def enrich_data(**context):
"""Étape 3: Enrichissement (embeddings)."""
print("🧠 Génération des embeddings...")
# Ici: appeler le code du Module 4
# from src.enrich import run_enrichment
# run_enrichment()
return {"n_embeddings": 950}
def deduplicate(**context):
"""Étape 3b: Déduplication sémantique."""
print("🔍 Déduplication...")
return {"n_duplicates": 20}
def generate_catalog(**context):
"""Étape 3c: Génération du catalogue."""
print("📋 Génération du catalogue...")
return {"catalog_path": "s3://processed/catalog.parquet"}
def generate_report(**context):
"""Étape 4: Rapport final."""
print("📊 Génération du rapport...")
# Récupérer les résultats des tâches précédentes
ti = context['ti']
ingest_result = ti.xcom_pull(task_ids='ingest')
print(f" Images ingérées: {ingest_result}")
return {"status": "success"}
# ============================================
# DÉFINIR LES TÂCHES
# ============================================
t_ingest = PythonOperator(
task_id='ingest',
python_callable=ingest_data,
dag=dag,
)
t_transform = PythonOperator(
task_id='transform',
python_callable=transform_data,
dag=dag,
)
t_enrich = PythonOperator(
task_id='enrich',
python_callable=enrich_data,
dag=dag,
)
t_dedup = PythonOperator(
task_id='deduplicate',
python_callable=deduplicate,
dag=dag,
)
t_catalog = PythonOperator(
task_id='catalog',
python_callable=generate_catalog,
dag=dag,
)
t_report = PythonOperator(
task_id='report',
python_callable=generate_report,
dag=dag,
)
# DVC push à la fin
t_dvc_push = BashOperator(
task_id='dvc_push',
bash_command='cd /opt/airflow/repo && dvc push',
dag=dag,
)
# ============================================
# DÉFINIR LES DÉPENDANCES
# ============================================
t_ingest >> t_transform >> [t_enrich, t_dedup, t_catalog] >> t_report >> t_dvc_push
'''
print("📋 DAG Airflow:")
print(dag_code)pythonVoir le code
# ============================================
# Sauvegarder le DAG dans un fichier
# ============================================
import os
# Créer le dossier dags
os.makedirs('dags', exist_ok=True)
# Sauvegarder le DAG
dag_content = '''
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
default_args = {
'owner': 'data-engineering',
'depends_on_past': False,
'start_date': datetime(2024, 1, 1),
'retries': 2,
'retry_delay': timedelta(minutes=5),
}
dag = DAG(
dag_id='data_engineering_pipeline',
default_args=default_args,
description='Pipeline DE for AI',
schedule_interval='@daily',
catchup=False,
)
def ingest_data(**context):
print("📥 Ingestion...")
return {"n_images": 1000}
def transform_data(**context):
print("🔄 Transform...")
return {"n_transformed": 950}
def enrich_data(**context):
print("🧠 Enrich...")
return {"n_embeddings": 950}
def generate_report(**context):
print("📊 Report...")
return {"status": "success"}
t_ingest = PythonOperator(task_id='ingest', python_callable=ingest_data, dag=dag)
t_transform = PythonOperator(task_id='transform', python_callable=transform_data, dag=dag)
t_enrich = PythonOperator(task_id='enrich', python_callable=enrich_data, dag=dag)
t_report = PythonOperator(task_id='report', python_callable=generate_report, dag=dag)
t_ingest >> t_transform >> t_enrich >> t_report
'''
with open('dags/data_pipeline_dag.py', 'w') as f:
f.write(dag_content)
print("✅ DAG sauvegardé: dags/data_pipeline_dag.py")
print("\n💡 Pour l'utiliser:")
print(" 1. Copie dags/ dans le dossier dags/ d'Airflow")
print(" 2. Redémarre Airflow ou attends le refresh")
print(" 3. Active le DAG dans l'UI http://localhost:8080")5.5 Pipeline complet intégré
Architecture finale
┌─────────────────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE COMPLÈTE │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────────────────────────────────────────────────────────────┐ │
│ │ AIRFLOW │ │
│ │ Orchestration: déclenche les tâches dans le bon ordre │ │
│ └──────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌────────────────────────┼────────────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ MinIO │ │ Scripts │ │ Chroma │ │
│ │ (données) │◄────────▶│ Python │◄────────▶│ (vectors) │ │
│ └─────────────┘ └──────┬──────┘ └─────────────┘ │
│ │ │
│ ┌───────────────────────┼───────────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ DVC │ │ MLflow │ │ Git │ │
│ │ (version │ │ (tracking) │ │ (code) │ │
│ │ données) │ │ │ │ │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
pythonVoir le code
# ============================================
# Pipeline intégré avec tracking
# ============================================
import time
from datetime import datetime
class IntegratedPipeline:
"""
Pipeline intégré:
- Exécute les étapes dans l'ordre
- Track tout avec MLflow
- Versionne avec DVC (optionnel)
"""
def __init__(self, config: dict):
self.config = config
self.results = {}
self.timings = {}
def _time_step(self, step_name: str, func):
"""Exécute une étape et mesure le temps."""
start = time.time()
result = func()
duration = time.time() - start
self.timings[step_name] = duration
self.results[step_name] = result
print(f" ✅ {step_name}: {duration:.1f}s")
return result
def step_ingest(self):
"""Simuler l'ingestion."""
time.sleep(0.5) # Simuler le travail
return {'n_images': 1000, 'n_rejected': 50}
def step_transform(self):
"""Simuler la transformation."""
time.sleep(0.3)
return {'n_transformed': 950, 'resize_mode': 'PAD'}
def step_enrich(self):
"""Simuler l'enrichissement."""
time.sleep(0.4)
return {'n_embeddings': 950, 'n_duplicates': 25}
def run(self) -> dict:
"""
Exécute le pipeline complet avec tracking MLflow.
"""
print("="*60)
print(f"🚀 PIPELINE INTÉGRÉ - {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print("="*60)
with DatasetTracker(run_name=f"pipeline-{datetime.now().strftime('%Y%m%d-%H%M%S')}") as tracker:
# 1. Log la config
print("\n📝 Configuration:")
tracker.log_pipeline_config(self.config)
# 2. Exécuter les étapes
print("\n🔄 Exécution:")
self._time_step('ingest', self.step_ingest)
self._time_step('transform', self.step_transform)
self._time_step('enrich', self.step_enrich)
# 3. Log les résultats
print("\n📊 Résultats:")
all_metrics = {}
for step, result in self.results.items():
for k, v in result.items():
if isinstance(v, (int, float)):
all_metrics[f"{step}_{k}"] = v
tracker.log_dataset_stats(all_metrics)
# 4. Log les timings
for step, duration in self.timings.items():
tracker.log_step_duration(step, duration)
total_time = sum(self.timings.values())
print(f"\n⏱️ Temps total: {total_time:.1f}s")
print("\n" + "="*60)
print("✅ Pipeline terminé ! Voir MLflow: http://localhost:5000")
print("="*60)
return self.results
print("✅ IntegratedPipeline défini")pythonVoir le code
# ============================================
# Exécuter le pipeline intégré
# ============================================
# Configuration
config = {
'source_bucket': 'raw-images',
'dest_bucket': 'processed-images',
'target_size': 256,
'resize_mode': 'PAD',
'normalize_range': '[-1, 1]',
'dedup_threshold': 0.95,
'version': 'v1.0'
}
# Exécuter
pipeline = IntegratedPipeline(config)
results = pipeline.run()5.6 Code source et ressources
Structure du projet final
data-engineering-for-ai/
├── docker-compose.yml # Tous les services
├── requirements.txt
├── dags/
│ └── data_pipeline_dag.py # DAG Airflow
├── src/
│ ├── ingestion.py # Module 2
│ ├── transform.py # Module 3
│ ├── enrich.py # Module 4
│ └── pipeline.py # Pipeline intégré
├── data/
│ └── .gitkeep
├── data.dvc # Pointeur DVC
├── .dvc/
│ └── config # Config DVC
└── .git/
pythonVoir le code
# ============================================
# Générer le package téléchargeable
# ============================================
import os
import zipfile
import base64
from IPython.display import HTML, display
# Créer la structure
os.makedirs('module5-automation/dags', exist_ok=True)
os.makedirs('module5-automation/src', exist_ok=True)
# requirements.txt
requirements = '''apache-airflow>=2.7.0
mlflow>=2.8.0
dvc>=3.30.0
dvc-s3>=3.0.0
boto3>=1.28.0
'''
with open('module5-automation/requirements.txt', 'w') as f:
f.write(requirements)
# docker-compose.yml
docker_compose = '''version: '3.8'
services:
minio:
image: minio/minio:latest
ports:
- "9000:9000"
- "9001:9001"
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin123
command: server /data --console-address ":9001"
volumes:
- minio_data:/data
airflow:
image: apache/airflow:2.7.0-python3.10
ports:
- "8080:8080"
environment:
- AIRFLOW__CORE__EXECUTOR=LocalExecutor
- AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=sqlite:////opt/airflow/airflow.db
- AIRFLOW__CORE__LOAD_EXAMPLES=False
volumes:
- ./dags:/opt/airflow/dags
command: standalone
mlflow:
image: ghcr.io/mlflow/mlflow:v2.8.0
ports:
- "5000:5000"
command: mlflow server --host 0.0.0.0 --port 5000
volumes:
- mlflow_data:/mlflow
volumes:
minio_data:
mlflow_data:
'''
with open('module5-automation/docker-compose.yml', 'w') as f:
f.write(docker_compose)
# DAG
with open('module5-automation/dags/data_pipeline_dag.py', 'w') as f:
f.write(dag_content)
# Créer le ZIP
zip_path = 'module5-automation.zip'
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, dirs, files in os.walk('module5-automation'):
for file in files:
file_path = os.path.join(root, file)
zipf.write(file_path, file_path)
# Lien de téléchargement
with open(zip_path, 'rb') as f:
b64 = base64.b64encode(f.read()).decode()
html = f'''
<a download="module5-automation.zip"
href="data:application/zip;base64,{b64}"
style="display: inline-block; padding: 10px 20px;
background-color: #4CAF50; color: white;
text-decoration: none; border-radius: 5px;
font-weight: bold;">
📦 Télécharger module5-automation.zip
</a>
'''
display(HTML(html))
print(f"\n✅ Archive créée: {zip_path}")🎯 Checkpoint — Quiz
Teste tes connaissances ! Réponds aux questions, puis déroule pour voir les réponses.
Question 1 : Pourquoi automatiser un pipeline de données ?
A) Parce que c'est plus joli
B) Pour éviter les oublis, les erreurs manuelles et rendre le process reproductible
C) Parce que c'est obligatoire
D) Pour impressionner le management
📖 Voir la réponse
Réponse : B
Un pipeline automatisé :
- Pas d'oubli d'étape : Tout s'exécute dans le bon ordre
- Retry automatique : Si une étape échoue, elle est relancée
- Reproductible : Mêmes inputs → mêmes outputs
- Historique : On sait qui a lancé quoi et quand
Question 2 : DVC stocke quoi dans Git ?
A) Les données elles-mêmes
B) Un pointeur (hash) vers les données stockées ailleurs
C) Les modèles
D) Rien, DVC n'utilise pas Git
📖 Voir la réponse
Réponse : B
DVC crée un fichier .dvc (ex: data.dvc) qui contient le hash des données. Ce petit fichier est versionné dans Git.
Les vraies données sont stockées dans un remote (MinIO, S3, GCS).
Git: data.dvc (1KB, contient le hash)
MinIO: ab/cd1234... (50GB, vraies données)
Question 3 : C'est quoi un DAG dans Airflow ?
A) Un fichier de configuration
B) Un graphe qui définit les tâches et leurs dépendances
C) Un type de base de données
D) Un format d'image
📖 Voir la réponse
Réponse : B
DAG = Directed Acyclic Graph (Graphe Orienté Acyclique)
- Directed : Les flèches vont dans un sens (A → B)
- Acyclic : Pas de boucle (A → B → A interdit)
- Graph : Des nœuds (tâches) et des arêtes (dépendances)
Ingest → Transform → Enrich → Report
Question 4 : MLflow sert à quoi pour le Data Engineer ?
A) Entraîner des modèles
B) Tracker les métadonnées du dataset (taille, paramètres, durées)
C) Remplacer Git
D) Faire du machine learning
📖 Voir la réponse
Réponse : B
Le DE utilise MLflow pour tracker :
- Paramètres : target_size, resize_mode, thresholds
- Métriques : n_images, n_rejected, success_rate
- Artifacts : catalog.parquet, rapports
- Durées : temps de chaque étape
Le ML Engineer utilise MLflow pour tracker les expériences de training (hyperparamètres, accuracy, modèles).
Question 5 : Comment restaurer la version v1.0 d'un dataset avec DVC ?
A) dvc download v1.0
B) git checkout v1.0 puis dvc checkout
C) dvc restore v1.0
D) Télécharger manuellement depuis S3
📖 Voir la réponse
Réponse : B
# 1. Checkout le code et le fichier .dvc de la version v1.0
git checkout v1.0
# 2. Télécharge les données correspondantes depuis le remote
dvc checkout
Le fichier data.dvc de la v1.0 contient le hash des données v1.0. DVC sait quoi télécharger.
Question 6 : Quel schedule Airflow exécute le DAG tous les jours à minuit ?
A) @hourly
B) @daily
C) @weekly
D) 0 0 * * *
📖 Voir la réponse
Réponses : B et D (les deux sont corrects !)
| Preset | Équivalent cron | Signification |
|---|---|---|
@hourly |
0 * * * * |
Toutes les heures |
@daily |
0 0 * * * |
Tous les jours à minuit |
@weekly |
0 0 * * 0 |
Tous les dimanches |
@monthly |
0 0 1 * * |
Le 1er de chaque mois |
@daily est un raccourci pour 0 0 * * *.
➡️ Prochaine étape
Module 6 : Projet final
Tu vas :
- Appliquer tout ce que tu as appris sur un vrai dataset
- Construire un pipeline end-to-end complet
- Livrer un dataset training-ready documenté
Module Data Engineering for AI — From Zero to Hero Bootcamp