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.

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).

À 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.