TP 11 – Le modèle de pourboire en temps réel, puis sous le capot

• 55 min · 13h25 – 14h20
• Notebook : jour3/TP11_ml_streaming_catalyst.ipynb
• Prérequis : out/modele_pourboire (TP 7) · simulateur courses

📁 Dans le dépôt

• Notebook : jour3/TP11_ml_streaming_catalyst.ipynb
• Données : scripts/stream_simulator.py + modèle out/modele_pourboire du TP 7
• Corrigé : jour3/corriges/tp11_ml_streaming_catalyst.py (publié en fin de J3)

À exécuter depuis la racine du dépôt. Pas encore cloné ? Voir Prérequis et installation.

Objectif : appliquer sur le flux de courses le modèle entraîné au TP 7, puis lire et comparer des plans d’exécution.

Réussi si : des prédictions sont produites en continu et vous savez expliquer le plan.

Avant de commencer

python scripts/stream_simulator.py courses --reset          # en direct
python scripts/stream_simulator.py courses --all --reset    # ou tout d'un coup

Si le modèle du TP 7 est absent, ou si vous y avez sauvegardé le LinearSVC, le notebook réentraîne le Random Forest en 1 minute : le TP a besoin de la colonne probability.

Étapes

1. La préparation des features, dans UNE fonction réutilisable en batch et en streaming

Reprenez les étapes du TP 7 : carte uniquement, heure, jointure avec zones pour Borough, manhattan, nuit, label (pourboire > 0).

2. Recharger le modèle du TP 7

PipelineModel.load("out/modele_pourboire").

3. Appliquer le modèle sur le flux

modele.transform(features(flux)), puis p_pourboire = vector_to_array("probability")[1].

4. Taux de pourboire prédit vs réel, par fenêtre de 30 min et par Borough

Prédit = moyenne de p_pourboire, réel = moyenne de label. Watermark de 10 minutes sur ts = F.col("tpep_pickup_datetime").cast("timestamp").

Note

Le watermark refuse le type TIMESTAMP_NTZ (sans fuseau) de nos données, d’où la conversion en timestamp.

Pourquoi la moyenne des probabilités et pas des prédictions 0/1 ?

5. foreachBatch en outputMode("append")

Pour chaque micro-batch, écrire en Parquet (out/tp11/suivi) et afficher le nombre de fenêtres. Le premier micro-batch écrit 0 fenêtre : aucune n’est encore close par le watermark. C’est normal.

6. Sous le capot : explain("formatted") sur la même requête, en CSV puis en Parquet

requete = lambda df: df.filter("trip_distance > 10").groupBy("PULocationID").agg(F.avg("total_amount"))

Comparez PushedFilters, ReadSchema et Batched. Surprise : Catalyst pousse le filtre et élague les colonnes dans les deux cas. Qu’est-ce qui diffère, alors ? Mesurez le temps de chaque version (moyenne de 3 exécutions).

7. AQE : le plan final n’existe qu’après l’exécution, sur le même objet

r = requete(pq)
r.collect()
r.explain()

Trouvez isFinalPlan=true et AQEShuffleRead coalesced. Combien de blocs *(n) (whole-stage codegen) ? Avec requete(pq).explain() vous obtiendriez un nouvel objet, encore isFinalPlan=false.

Bonus – Spark UI, onglet Structured Streaming

Débit d’entrée, durée des micro-batches, lignes d’état. Process Rate > Input Rate : le flux tient la charge.

Correction

import os, shutil, time
from pyspark.sql import functions as F
from pyspark.ml import PipelineModel
from pyspark.ml.functions import vector_to_array
spark.conf.set("spark.sql.shuffle.partitions", "4")
for d in ["out/checkpoints/tp11", "out/tp11"]:
    shutil.rmtree(d, ignore_errors=True)
zones = spark.read.csv("data/taxi_zone_lookup.csv", header=True, inferSchema=True) \
             .select(F.col("LocationID").alias("PULocationID"), "Borough")

# 1. La MÊME préparation en batch et en streaming
def features(df):
    return (df.filter("trip_distance > 0 AND payment_type = 1")
              .withColumn("heure", F.hour("tpep_pickup_datetime"))
              .join(zones, "PULocationID")
              .withColumn("manhattan", (F.col("Borough") == "Manhattan").cast("double"))
              .withColumn("nuit", ((F.col("heure") >= 20) | (F.col("heure") < 5)).cast("double"))
              .withColumn("label", (F.col("tip_amount") > 0).cast("double")))

# 2. Recharger (ML persistence)
modele = PipelineModel.load("out/modele_pourboire")

# 3. transform() fonctionne sur un DataFrame en streaming
schema = spark.read.parquet("data/trips.parquet").schema
flux = spark.readStream.schema(schema).option("maxFilesPerTrigger", 3).json("stream/courses")
predictions = (modele.transform(features(flux))
               .withColumn("p_pourboire", vector_to_array("probability")[1])
               .select("tpep_pickup_datetime", "Borough", "label", "prediction", "p_pourboire"))

# 4. Prédit vs réel par fenêtre de 30 minutes et par arrondissement
suivi = (predictions
         .withColumn("ts", F.col("tpep_pickup_datetime").cast("timestamp"))
         .withWatermark("ts", "10 minutes")
         .groupBy(F.window("ts", "30 minutes"), "Borough")
         .agg(F.count("*").alias("courses"),
              F.round(F.avg("p_pourboire"), 3).alias("taux_predit"),   # moyenne des probabilités
              F.round(F.avg("label"), 3).alias("taux_reel")))

# 5. foreachBatch : du code batch libre pour chaque micro-batch
def traiter(batch_df, batch_id):
    batch_df.write.mode("append").parquet("out/tp11/suivi")
    n = batch_df.count()
    if n:
        print(f"micro-batch {batch_id} : {n} fenêtres closes écrites")

q = (suivi.writeStream.outputMode("append").foreachBatch(traiter)
     .option("checkpointLocation", "out/checkpoints/tp11")
     .trigger(processingTime="5 seconds").start())
q.awaitTermination(120)
q.stop()

res = spark.read.parquet("out/tp11/suivi")
res.filter("Borough = 'Manhattan'") \
   .select(F.date_format("window.start", "dd/MM HH:mm").alias("fenetre"), "courses", "taux_predit", "taux_reel") \
   .orderBy("fenetre").show(12)
print("Écart moyen |prédit - réel| :",
      res.select(F.round(F.avg(F.abs(F.col("taux_predit") - F.col("taux_reel"))), 3)).first()[0])

# 6. Catalyst : même plan logique quelle que soit la source
requete = lambda df: df.filter("trip_distance > 10").groupBy("PULocationID").agg(F.avg("total_amount"))
csv = spark.read.csv("data/trips.csv", header=True, schema=schema)
pq = spark.read.parquet("data/trips.parquet")
requete(csv).explain("formatted")
requete(pq).explain("formatted")
for nom, df in [("CSV", csv), ("Parquet", pq)]:
    t = time.time()
    for _ in range(3):
        requete(df).collect()
    print(f"{nom:8s} : {(time.time() - t) / 3:.2f} s par requête")

# 7. AQE : le plan est réoptimisé PENDANT l'exécution
print("AQE activé :", spark.conf.get("spark.sql.adaptive.enabled"))
r = requete(pq)
r.collect()
r.explain()     # isFinalPlan=true, AQEShuffleRead coalesced

# 8. Tungsten : blocs *(n) compilés ensemble
import re
plan_final = r._jdf.queryExecution().executedPlan().toString()
print("Blocs whole-stage codegen :", len(set(re.findall(r"\*\((\d+)\)", plan_final))))

Résultats attendus

Mesure	Valeur
Prédictions	Écrites à chaque micro-batch (le premier écrit 0 fenêtre)
Écart moyen prédit / réel	≈ 0,21 (fenêtres de 30 à 70 courses : du bruit)
CSV / Parquet	≈ 0,4 s / ≈ 0,15 s par requête : × 2 à 3
Plan final	isFinalPlan=true, AQEShuffleRead coalesced
Whole-stage codegen	2 blocs

Pièges rencontrés en séance

Symptôme	Cause	Solution
probability introuvable	LinearSVC sauvegardé au TP 7	Réentraîner le Random Forest (cellule prévue)
Erreur de watermark sur tpep_pickup_datetime	Type TIMESTAMP_NTZ	.cast("timestamp") avant withWatermark
Premier micro-batch : 0 fenêtre	Aucune fenêtre close	Normal, attendre
isFinalPlan=false	explain() sur un nouvel objet	r.collect() puis r.explain() sur le même r
« Catalyst ne fait rien en CSV »	Mauvaise lecture du plan	PushedFilters est bien là : la différence est physique

Partage

• Un participant présente son suivi et son plan.
• Pourquoi Parquet est-il 2 à 3 fois plus rapide si Catalyst pousse le filtre dans les deux cas ? Format colonne, statistiques min/max, lecture vectorisée.
• Comment réduire le bruit du taux prédit ? Fenêtres plus longues, agréger plusieurs quartiers, intervalle de confiance.
• À retenir : même code, même plan. Le format et le moteur font la performance.