Partie 2 – Spark ML : pipelines, régression et persistance

• 10h45 – 12h00
• Théorie 25 min · TP 6 30 min · partage et persistance 20 min
• Slides 14 à 23

Et si toute la préparation et le modèle tenaient dans un seul objet, prêt pour la production ?

Dans beaucoup de projets, la préparation des données en production finit par différer de celle de l’entraînement. Le Pipeline règle ce problème.

Transformer, Estimator, Model : les 3 briques

Brique	Méthode	Rôle	Exemples
Transformer	transform(df) → df	Transforme un DataFrame en un autre, sans rien apprendre	VectorAssembler, un modèle entraîné
Estimator	fit(df) → Model	Apprend à partir des données	StringIndexer, OneHotEncoder, DecisionTreeRegressor
Model	transform(df) → df	Le résultat de fit(). C’est un Transformer qui sait prédire	model.transform(test) ajoute la colonne prediction

 train ──► Estimator.fit() ──► Model
                                 │
 test  ──────────────────────► Model.transform() ──► test + prediction

Analogie : l’Estimator est l’élève qui révise, fit() est l’examen, le Model est l’élève diplômé qui sait répondre.

Deux pièges de vocabulaire

• StringIndexer et OneHotEncoder sont des Estimators : ils apprennent la liste des catégories. VectorAssembler est un simple Transformer.
• Tout vient de pyspark.ml (API DataFrames). L’ancienne pyspark.mllib (RDD) est en maintenance depuis Spark 2.0 : à ne plus utiliser.

Spark ML veut toutes les features dans une seule colonne vecteur, nommée features par défaut : c’est le rôle de VectorAssembler.

Le Pipeline : enchaîner les étapes

 StringIndexer      OneHotEncoder      VectorAssembler        DecisionTreeRegressor
 (Estimator)        (Estimator)        (Transformer)          (Estimator)
 Borough → index    index → vecteur    colonnes → features    apprend fare_amount
       └──────────────────┴──────────────────┴────────────────────────┘
                     Pipeline(stages=[…]) – lui-même un Estimator

 pipeline.fit(train)   → PipelineModel : chaque Estimator est remplacé par son Model
 model.transform(test) → prédictions : toutes les étapes rejouées, dans le même ordre

Pourquoi le Pipeline :

• Mêmes transformations à l’entraînement et en production : aucun décalage possible.
• Un seul objet à sauvegarder, versionner et déployer.
• Pas de fuite : les statistiques (catégories, moyennes) sont apprises sur le train uniquement.
• Comparer des modèles = changer seulement la dernière étape.

Pourquoi le one-hot ? Un index 0, 1, 2 laisserait croire que Brooklyn (2) est « plus grand » que Manhattan (0). Le vecteur 0/1 supprime cet ordre artificiel.

🔍 Vulgarisation : StringIndexer + OneHotEncoder + VectorAssembler expliqués pas à pas sur 3 lignes

Suivons comment une ligne de course se transforme à travers les 3 étapes de préparation du Pipeline :

Étape	Entrée	Sortie ajoutée au DataFrame	Pourquoi cette étape est indispensable ?
Départ	Borough = "Queens", trip_distance = 18.0, heure = 14	—	Les algorithmes mathématiques ne savent pas multiplier le mot "Queens" par un coefficient !
1. StringIndexer	"Manhattan" (fréquent), "Queens", "Brooklyn"	borough_idx = 1.0 (0 = Manhattan, 1 = Queens, 2 = Brooklyn)	Remplace le texte par un numéro. Mais attention : pour une régression linéaire, 2 est le double de 1, et la moyenne de Manhattan (0) et Brooklyn (2) vaudrait Queens (1), ce qui n’a aucun sens géographique !
2. OneHotEncoder	borough_idx = 1.0	borough_vec = [0.0, 1.0, 0.0] (un interrupteur 0/1 par quartier)	Chaque quartier devient une coordonnée indépendante (Manhattan oui/non, Queens oui/non…). Plus aucun faux rapport d’ordre !
3. VectorAssembler	18.0, 14, [0.0, 1.0, 0.0]	features = [18.0, 14.0, 0.0, 1.0, 0.0]	Spark ML exige que toutes les variables explicatives soient regroupées dans une seule colonne de type Vector.

Et que fait handleInvalid="keep" sur StringIndexer ? Si dans votre jeu de train (80 %) aucune course ne partait de "Staten Island", mais qu’une course en part dans le jeu de test (20 %) ou en production demain, sans handleInvalid="keep", Spark plante avec une erreur Unseen label: Staten Island. Avec "keep", il range proprement cette catégorie nouvelle dans un index réservé aux inconnus.

L’arbre de régression : des questions en cascade

                    distance ≤ 6 miles ?
                    /                 \
                 oui                   non
                 /                       \
     distance ≤ 2,5 miles ?        départ du Queens ?
        /          \                 /          \
     oui            non           oui            non
    ≈ 11 $         ≈ 22 $        ≈ 70 $         ≈ 57 $

• Chaque nœud pose une question sur une feature. Chaque feuille prédit la moyenne des courses qui y arrivent.
• maxDepth règle la profondeur : trop profond = sur-apprentissage.
• L’arbre capte les ruptures (forfait aéroport JFK) qu’une droite ne voit pas.

Suivez une course de 3 miles depuis Midtown : distance ≤ 6 ? oui ; ≤ 2,5 ? non → environ 22 $. À droite, au-delà de 6 miles, partir du Queens (JFK, LaGuardia) signifie souvent le forfait aéroport de 70 $ : l’arbre retrouve le forfait sans connaître RatecodeID.

Le schéma est simplifié. L’arbre réel de profondeur 6 a 127 nœuds ; son premier découpage est bien « distance ≤ 5,96 miles », à montrer en TP avec model.stages[-1].toDebugString. Le palier à 70 $ du nuage de points de la partie 1 est exactement ce que l’arbre capture.

En code : prédire le montant d’une course

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml.regression import DecisionTreeRegressor
from pyspark.ml.evaluation import RegressionEvaluator

train, test = df.randomSplit([0.8, 0.2], seed=42)
pipeline = Pipeline(stages=[
    StringIndexer(inputCol="Borough", outputCol="borough_idx", handleInvalid="keep"),
    OneHotEncoder(inputCols=["borough_idx"], outputCols=["borough_vec"]),
    VectorAssembler(inputCols=["trip_distance", "heure", "passenger_count", "borough_vec"],
                    outputCol="features"),
    DecisionTreeRegressor(featuresCol="features", labelCol="fare_amount", maxDepth=6),
])
model = pipeline.fit(train)               # apprend sur 80 % des courses
pred  = model.transform(test)             # prédit sur 20 % jamais vues
rmse  = RegressionEvaluator(labelCol="fare_amount", metricName="rmse").evaluate(pred)

heure est créée avant avec F.hour("tpep_pickup_datetime") ; Borough vient de la jointure avec taxi_zone_lookup.csv.

Modèle (test, 20 %)	RMSE	R²
Régression linéaire	5,34 $	0,914
Arbre, maxDepth = 6	4,66 $	0,935

• RMSE : l’erreur typique, dans l’unité du label (ici en dollars). Elle pénalise les grosses erreurs.
• R² : part de la variation expliquée. 1 = parfait, 0 = pas mieux que la moyenne.

L’arbre se trompe en moyenne de 4,66 $ contre 5,34 $ pour la régression linéaire : la droite ne sait pas modéliser le forfait JFK à 70 $. Résultats réels sur le fil rouge (PySpark 4.x).

🧮 Calcul à la main : Comprendre concrètement RMSE et R² sur 3 courses de taxi

Prenons 3 courses du jeu de test (dont la moyenne des vrais tarifs est $\bar{y} = 30$$) :

Course	Vrai tarif ($y$)	Prédiction du modèle ($\hat{y}$)	Erreur ($y - \hat{y}$)	Erreur au carré $(y - \hat{y})^2$
Course A (courte)	10 $	12 $	−2 $	$4$
Course B (moyenne)	20 $	18 $	+2 $	$4$
Course C (forfait JFK)	60 $	55 $	+5 $	$25$

1. Calcul de la RMSE (Root Mean Squared Error) :
   • Moyenne des carrés des erreurs (MSE) : $\frac{4 + 4 + 25}{3} = \frac{33}{3} = 11\text{ $}^2$.
   • Racine carrée pour revenir en dollars (RMSE) : $\sqrt{11} \approx$ 3,32 $.
   • Pourquoi élever au carré avant de faire la moyenne ? D’abord pour que $-2$$ et $+2$$ ne s’annulent pas ($0$$), et ensuite parce que le carré punit sévèrement les grosses erreurs : une seule erreur de $10$$ ($10^2 = 100$) pèse autant que 25 erreurs de $2$$ ($25 \times 4 = 100$) !
2. Calcul du $R^2$ (coefficient de détermination) :
   • Compare l’erreur de votre modèle à celle d’un modèle « bête » qui répondrait toujours le tarif moyen ($30$$).
   • $R^2 = 0,935$ signifie que votre arbre de décision explique 93,5 % des écarts de prix entre les courses.

À vous de jouer

TP 6 – Prédire le montant (30 min) : régression linéaire contre arbre, RMSE et R² sur le test. Réussi si la RMSE de l’arbre est inférieure à 5 $.

Partage et remédiation

• Pourquoi l’arbre bat-il la régression linéaire ? Il capte le palier du forfait JFK (une rupture), la droite non.
• Bonus : R² = 0,976 avec total_amount. Bonne nouvelle ? Non. total_amount = tarif + suppléments + pourboire : il contient déjà la réponse et n’est connu qu’après la course. C’est une fuite de données, le piège n° 1 du ML, repris en partie 3.
• À retenir : une feature inconnue au moment de prédire = fuite de données.

Sauvegarder un modèle et changer de langage

from pyspark.ml import PipelineModel

model.write().overwrite().save("out/modele_montant")      # préparation + modèle, tout le pipeline
recharge = PipelineModel.load("out/modele_montant")
recharge.transform(test.limit(5)).select("total_amount", "prediction").show()   # test : le jeu réservé plus haut

L’arborescence créée contient metadata et stages. Le format est portable : entraîné en Python, rechargé en Scala ou Java. Cas d’usage : entraîner chaque nuit sur le cluster, recharger le modèle dans un job de prédiction (batch, ou streaming demain au TP 11).

Langage	Situation
PySpark	L’API de référence, celle de nos TP
Scala, Java	Même API spark.ml, mêmes noms de classes
SparkR	Déprécié depuis Spark 4.0, « will be removed in a future version »
R avec sparklyr	L’alternative maintenue pour les équipes R (projet Posit, syntaxe dplyr)

PySpark ou R ?

Les TP s’écrivent en PySpark, l’API la plus utilisée en entreprise et celle qui reçoit toutes les nouveautés de Spark 4. Les équipes R trouveront leur équivalent dans sparklyr (projet Posit, syntaxe dplyr), maintenu et recommandé : SparkR, livré avec Spark, est déprécié depuis la 4.0.

Avant le déjeuner : reprise à 13h00 avec la partie 3.