Partie 3 – Sous le capot (Catalyst, Tungsten, AQE) et Machine Learning en temps réel

• 13h00 – 14h45
• Théorie 25 min · TP 11 55 min · partage 15 min
• Slides 24 à 30

Et si le même calcul pouvait aller 2 à 3 fois plus vite sans changer une ligne de logique ? Chiffre réel du fil rouge : la même requête prend ≈ 0,4 s sur le CSV et ≈ 0,15 s sur le Parquet.

Catalyst : de votre code au plan d’exécution

 Votre code        Plan logique    Plan logique    Plan logique    Plan physique    Code généré
 SQL, Python,  ──► non résolu  ──► résolu      ──► optimisé    ──► choisi       ──► Tungsten
 Scala
                        │ Analyse        │ Optimisation      │ Planification
                        │ le catalogue   │ règles : filtres  │ comparer les stratégies
                        │ vérifie        │ poussés, colonnes │ (broadcast ou
                        │ colonnes/types │ élaguées,         │ sort-merge join)
                                         │ constantes

Le plan logique décrit quoi calculer. Catalyst le réécrit avec des règles (predicate pushdown, column pruning, constant folding…), puis choisit comment (plan physique) selon les coûts : broadcast join si une table est petite, ce qui se passe avec taxi_zone_lookup.csv.

Ce que montre explain("formatted") sur le fil rouge, pour df.filter("trip_distance > 10").groupBy("PULocationID").agg(F.avg("total_amount")) :

(1) Scan parquet
    Batched: true
    PushedFilters: [IsNotNull(trip_distance), GreaterThan(trip_distance,10.0)]
    ReadSchema: struct<trip_distance:double,PULocationID:int,total_amount:double>

Ligne	Ce qu’elle dit
PushedFilters	Le filtre est appliqué pendant la lecture, pas après
ReadSchema	3 colonnes sur 19 : les autres ne sont jamais lues
Batched: true	Lecture vectorisée (Tungsten)

Même plan, même performance en Python, Scala ou SQL : Python ne fait que construire le plan, l’exécution se passe dans la JVM. Exceptions : les UDF Python (sérialisation ligne à ligne ; préférer les pandas UDF vectorisées) et les lambdas typées, opaques pour Catalyst.

Tungsten et AQE : exécuter vite, corriger en route

Tungsten : le moteur d’exécution	AQE : l’optimisation adaptative
Données en binaire compact, hors des objets Java (pas de ramasse-miettes sur des millions d’objets)	Le plan est réoptimisé pendant l’exécution, avec les vraies tailles
Whole-stage codegen : les opérateurs *(n) sont compilés ensemble en une seule boucle	Fusionne les petites partitions de shuffle (AQEShuffleRead coalesced)
Lecture vectorisée (Batched: true)	Bascule en broadcast join, découpe les partitions déséquilibrées (skew)

AQE est activé par défaut depuis Spark 3.2. Après chaque shuffle, Spark connaît la vraie taille des données et ajuste le plan.

Mesuré sur le fil rouge (même requête, 3 exécutions, 120 000 courses)	Temps par requête
CSV	≈ 0,4 s (0,38 à 0,42)
Parquet	≈ 0,15 s (0,14 à 0,15)
Gain	× 2 à 3

Les chiffres varient selon le poste : retenir l’ordre de grandeur, et un écart qui grandit avec le volume.

Le piège

Catalyst pousse le filtre dans les deux cas : PushedFilters apparaît aussi en CSV. La différence vient du format physique : Parquet est en colonnes (on ne lit que 3 colonnes), compressé, avec des statistiques min/max qui permettent de sauter des blocs, et lu en vectorisé. Le CSV doit lire et découper chaque ligne entière.

🔍 Vulgarisation : Catalyst, Tungsten et AQE expliqués comme un GPS, un moteur de Formule 1 et un recalcul d'itinéraire en direct

1. Catalyst = le planificateur d’itinéraire (avant le départ) :
   • Vous écrivez une requête maladroite : « Joins les deux tables, garde toutes les colonnes, puis filtre sur trip_distance > 10 ».
   • Catalyst réécrit votre plan avant de démarrer : « Je vais filtrer trip_distance > 10 D’ABORD pendant la lecture, je ne vais charger que les 3 colonnes utiles sur 19, et je vais envoyer la petite table des zones en Broadcast ».
2. Tungsten = le moteur de Formule 1 (pendant la course) :
   • Au lieu de créer des millions d’objets Java en mémoire (qui ralentissent le Garbage Collector), Tungsten range les nombres directement en binaire compact dans la RAM et fusionne (Whole-Stage Codegen *(1)) le filtre, la projection et la somme dans une seule boucle while en bytecode Java ultra-rapide.
3. AQE (Adaptive Query Execution) = le GPS Waze qui change de route quand il voit un bouchon :
   • Avant de lancer le job, Catalyst ne sait pas combien de lignes survivront à votre filtre trip_distance > 10.
   • Avec AQE, Spark s’arrête un instant au moment du shuffle, pèse les vraies données en mémoire et se dit : « Il ne reste que 50 Ko de données ! Pas besoin de 200 partitions de shuffle, je les fusionne en 2 partitions (AQEShuffleRead coalesced), et je bascule la jointure en Broadcast Join à la volée ! »

Le modèle d’hier, appliqué au flux

 Courses      features()        PipelineModel      Fenêtres        foreachBatch
 en direct ─► même code    ─►   du TP 7       ─►  30 min     ─►   Parquet, alertes
              qu'en batch

Extrait du corrigé jour3/corriges/tp11_ml_streaming_catalyst.py ; features() est la fonction du TP 7 (filtre carte, heure, jointure zones, manhattan, nuit, label), reprise à l’identique.

from pyspark.ml import PipelineModel
from pyspark.ml.functions import vector_to_array

modele = PipelineModel.load("out/modele_pourboire")               # le modèle entraîné au TP 7
schema = spark.read.parquet("data/trips.parquet").schema           # un flux exige un schéma explicite
flux   = spark.readStream.schema(schema).json("stream/courses")

pred = (modele.transform(features(flux))                           # transform() marche aussi sur un flux
        .withColumn("p_pourboire", vector_to_array("probability")[1])   # probabilité de la classe « pourboire »
        .withColumn("ts", F.col("tpep_pickup_datetime").cast("timestamp")))  # le watermark refuse TIMESTAMP_NTZ

suivi = (pred.withWatermark("ts", "10 minutes")
             .groupBy(F.window("ts", "30 minutes"), "Borough")
             .agg(F.avg("p_pourboire").alias("taux_predit"), F.avg("label").alias("taux_reel")))

def traiter(batch_df, batch_id):                                   # du code batch libre à chaque micro-batch
    batch_df.write.mode("append").parquet("out/tp11/suivi")

q = (suivi.writeStream.outputMode("append").foreachBatch(traiter)
     .option("checkpointLocation", "out/checkpoints/tp11")         # obligatoire pour une agrégation
     .trigger(processingTime="5 seconds")
     .start())

• Une seule fonction features() pour l’entraînement et la production : aucun décalage. Sinon, le modèle reçoit en production des données qu’il n’a jamais vues (training-serving skew).
• foreachBatch : chaque micro-batch est un DataFrame ordinaire (écrire, fusionner, appeler une API).
• Taux prédit = moyenne des probabilités (p_pourboire), pas des prédictions 0/1 : la moyenne de prédictions à seuil 0,5 surestime systématiquement.
• Prédit vs réel par fenêtre : écart moyen 0,21. Les fenêtres de 30 minutes par quartier contiennent peu de courses (30 à 70), donc beaucoup de bruit.

🧮 Démonstration mathématique : Pourquoi faut-il moyenner p_pourboire (probabilité) et JAMAIS prediction (0/1) ?

Supposons que dans une fenêtre de 30 minutes il y ait 10 courses, et que pour chacune le modèle Random Forest estime 60 % de chance de pourboire (p_pourboire = 0.60). En réalité, sur ces 10 courses à 60 % de chance, 6 clients donnent un pourboire (taux_reel = 0.60).

Que se passe-t-il si vous calculez votre taux_predit des deux manières ?

1. ❌ Si vous faites la moyenne de la colonne prediction (0 ou 1 avec seuil 0,5) :
   • Comme $0,60 \ge 0,50$, prediction vaut 1.0 pour chacune des 10 courses !
   • La moyenne des prediction vaut $\frac{1+1+\dots+1}{10} =$ 1,00 (100 % de pourboires prédits) au lieu de 60 % ! Votre indicateur surestime massivement la réalité.
2. ✅ Si vous faites la moyenne de la probabilité p_pourboire (F.avg("p_pourboire")) :
   • La moyenne des probabilités vaut $\frac{0,60 + 0,60 + \dots + 0,60}{10} =$ 0,60 (60 %), ce qui correspond exactement à l’espérance mathématique du nombre de pourboires (taux_reel = 0.60) !

La courbe à surveiller en production

Spark UI, onglet Structured Streaming : Input Rate ≈ 250 lignes/s, Process Rate 600 à 1 000 lignes/s. Tant que Process Rate > Input Rate, le flux tient la charge. Sinon la file s’allonge et la latence explose.

C’est le cliffhanger annoncé hier : le même PipelineModel, sauvegardé au TP 7, sert en batch et en streaming.

À vous de jouer

TP 11 – ML en temps réel et plans (55 min) : le modèle du jour 2 sur le flux, puis sous le capot avec explain.

Partage et remédiation

• Pourquoi Parquet est-il 2 à 3 fois plus rapide si Catalyst pousse le filtre dans les deux cas ? Format colonne compressé, statistiques min/max, lecture vectorisée. Le CSV lit et découpe chaque ligne.
• Comment réduire le bruit du taux prédit par fenêtre ? Fenêtres plus longues, agréger plusieurs quartiers, ou afficher un intervalle de confiance.
• À retenir : même code, même plan. Le format et le moteur font la performance.

Pause de 15 minutes à 14h45, puis la partie 4. Le QCM est à 16h.