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.

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.

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.