Fiche mémo – Jour 3

En une minute

Un flux, c’est une table qui grandit sans fin : on l’écrit presque comme en batch, avec readStream et writeStream à la place de read et write. Le checkpoint est la mémoire de la requête : il permet de reprendre après une panne, et c’est lui qui fait ignorer les fichiers déjà lus. Le watermark est LA notion du jour : il dit jusqu’où attendre les retardataires, ce qui permet de clore les fenêtres et de libérer l’état. Un modèle Spark ML sauvegardé s’applique tel quel sur un flux avec transform(), et foreachBatch rend tout le code batch pour chaque micro-batch. Sous le capot, Catalyst optimise le plan quelle que soit la source, Tungsten compile les opérateurs, AQE réajuste le plan pendant l’exécution, et le format Parquet fait le reste.

Notions clés

Notion	Définition en une ligne	Piège fréquent
readStream	Lit une source en continu et renvoie un DataFrame avec isStreaming = True.	Oublier le schéma : un stream de fichiers exige un schéma explicite.
Micro-batch / trigger	Spark traite les nouvelles données par paquets : processingTime, availableNow=True, realTime (mode temps réel de Spark 4.1 ; mot-clé PySpark depuis 4.2).	Croire que chaque fichier fait un micro-batch : c’est maxFilesPerTrigger qui fixe le nombre de fichiers par paquet.
Checkpoint	Dossier qui mémorise les fichiers lus, l’état et la progression : reprise après panne.	Relancer après un --reset du simulateur sans supprimer le checkpoint : la requête reprend l’ancien état.
Sources / sinks	Fichiers, Kafka (rejouable) / fichiers, Kafka, foreachBatch, console, memory.	Une source non rejouable ou un sink non idempotent : pas d’exactly-once.
Jointure flux-statique	Enrichir un flux avec un DataFrame ordinaire (les zones et leur Borough).	Croire qu’il faut transformer le référentiel en stream : c’est un spark.read classique.
Output mode complete	Réécrit toute la table résultat à chaque micro-batch.	Possible seulement avec une agrégation ; l’état grossit sans limite.
Output mode update	N’émet que les lignes modifiées depuis le dernier micro-batch.	Le confondre avec append : une même fenêtre peut sortir plusieurs fois.
Output mode append	N’émet une ligne qu’une fois, quand elle est définitive.	Sur une agrégation, il faut un watermark : sans fenêtre close, rien ne sort.
Fenêtre temporelle	F.window("ts", "1 minute") regroupe selon le temps de l’événement ; glissante avec un pas ; session_window.	Grouper sur l’heure de traitement : les retards faussent les comptes.
Watermark	« Une ligne qui arrive avec plus de X de retard est ignorée » ; Spark peut alors clore les fenêtres.	S’étonner que les dernières minutes manquent dans le Parquet : elles ne sont pas encore closes.
Non supporté en streaming	Le count distinct exact et le tri sans agrégation.	Chercher un bug : il faut une version approchée (approx_count_distinct, percentile_approx) ou un batch.
foreachBatch	Exécute une fonction (batch_df, batch_id) avec du code batch libre sur chaque micro-batch.	S’inquiéter du premier micro-batch vide en append : aucune fenêtre n’est encore close.
Encoders / Datasets	En Scala et Java, Dataset[T] typé ; l’Encoder traduit l’objet en binaire Tungsten. Python : DataFrame + schéma.	Les lambdas typées sont opaques pour Catalyst : préférer F.col.
Catalyst	Plan logique → optimisé → physique : pushdown du filtre, élagage des colonnes, choix du join.	Croire que le pushdown n’existe qu’en Parquet : PushedFilters apparaît aussi en CSV.
Tungsten	Binaire compact, whole-stage codegen (blocs *(n)), lecture vectorisée (Batched: true).	Attribuer à Catalyst le gain de Parquet : la différence vient du format.
AQE	Réoptimise le plan pendant l’exécution : fusion des petites partitions, broadcast, skew. Par défaut depuis 3.2.	Appeler explain() sur un nouvel objet : le plan reste isFinalPlan=false.
Flume (héritage)	Agent = source + channel + sink ; Avro RPC entre agents. Connecteur Spark retiré en 3.0.	Le prendre pour une technologie actuelle : aujourd’hui Kafka, et Avro dans Kafka.
DStreams (héritage)	StreamingContext, un RDD par micro-batch, temps de traitement seulement. Legacy.	Écrire un nouveau projet en DStreams.
Deep Learning avec Spark	binaryFile, predict_batch_udf, pandas UDF, TorchDistributor. Deep Learning Pipelines retiré.	Attendre de Spark qu’il entraîne lui-même un réseau de neurones.
SparkR / sparklyr	SparkR déprécié en 4.0 ; sparklyr (Posit, dplyr) pour R.	Lancer un projet R sur SparkR.
Lambda / Kappa	Lambda : chaîne batch + chaîne flux ; Kappa : un seul flux, journal rejouable.	Choisir par mode plutôt que par besoin : tableau de bord, alertes ou facturation exacte.

Code à connaître

Des extraits, pas des programmes complets : chacun suppose les imports et les variables du corrigé indiqué en commentaire (jour3/corriges/ dans le dépôt, notebooks TP9_premier_stream, TP10_logs_temps_reel, TP11_ml_streaming_catalyst et scripts .py du même nom). Imports communs :

from pyspark.sql import functions as F
from pyspark.ml import PipelineModel
from pyspark.ml.functions import vector_to_array

# tp9_premier_stream.py – lire un dossier en continu : schéma explicite, repris des données statiques
schema = spark.read.parquet("data/trips.parquet").schema
courses = (spark.readStream.schema(schema)
           .option("maxFilesPerTrigger", 2)       # 2 fichiers par micro-batch
           .json("stream/courses"))

# tp9_premier_stream.py – jointure flux-statique, agrégation continue, sortie complete
zones = spark.read.csv("data/taxi_zone_lookup.csv", header=True, inferSchema=True)
par_borough = (courses.join(zones.select(F.col("LocationID").alias("PULocationID"), "Borough"), "PULocationID")
               .groupBy("Borough").agg(F.count("*").alias("courses"), F.sum("total_amount").alias("ca")))
q = (par_borough.writeStream.outputMode("complete").format("console")
     .trigger(processingTime="5 seconds")
     .option("checkpointLocation", "out/checkpoints/tp9")   # un checkpoint par requête
     .start())

# tp10_logs_temps_reel.py – fenêtre de 1 minute sur le temps de l'événement, watermark de 2 minutes, sortie append
# (logs : les lignes parsées, avec ts en timestamp, endpoint et status)
par_minute = (logs.withWatermark("ts", "2 minutes")
              .groupBy(F.window("ts", "1 minute"), "endpoint")
              .agg(F.count("*").alias("requetes"),
                   F.sum(F.when(F.col("status") >= 500, 1).otherwise(0)).alias("erreurs")))

# tp10_logs_temps_reel.py – suivre la requête : lignes lues, lignes ignorées par le watermark
q.recentProgress[-1]["numInputRows"]
q.recentProgress[-1]["stateOperators"][0]["numRowsDroppedByWatermark"]

# tp11_ml_streaming_catalyst.py – modèle ML sur un flux : même fonction features() en batch et en streaming
modele = PipelineModel.load("out/modele_pourboire")
predictions = (modele.transform(features(flux))
               .withColumn("p_pourboire", vector_to_array("probability")[1])
               .withColumn("ts", F.col("tpep_pickup_datetime").cast("timestamp")))   # le watermark refuse TIMESTAMP_NTZ
suivi = (predictions.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")))

# tp11_ml_streaming_catalyst.py – foreachBatch : du code batch libre pour chaque micro-batch
def traiter(batch_df, batch_id):
    batch_df.write.mode("append").parquet("out/tp11/suivi")
q = suivi.writeStream.outputMode("append").foreachBatch(traiter) \
         .option("checkpointLocation", "out/checkpoints/tp11").start()

# tp11_ml_streaming_catalyst.py – lire un plan ; AQE : le plan final n'existe qu'après exécution, sur le même objet
requete = lambda df: df.filter("trip_distance > 10").groupBy("PULocationID").agg(F.avg("total_amount"))
pq = spark.read.parquet("data/trips.parquet")
requete(pq).explain("formatted")                # PushedFilters, ReadSchema, Batched
r = requete(pq); r.collect(); r.explain()       # isFinalPlan=true, AQEShuffleRead coalesced

Chiffres des TP

Valeur	Contexte
1 fichier / 2 s	Rythme du simulateur en direct ; --all dépose tout d’un coup
4	spark.sql.shuffle.partitions, au lieu des 200 par défaut
2 fichiers, 5 s, 60 s	TP 9 : maxFilesPerTrigger, trigger, durée de la démonstration
14 992 courses, 15 micro-batches	TP 9 avec le flux complet : Manhattan 11 880, Queens 1 553…
3 fichiers, 1 min, 2 min	TP 10 : maxFilesPerTrigger, fenêtre, watermark
/payment, 8h22–8h25	TP 10 : la panne. Taux d’erreur de ≈ 1 % à 29–38 %, p95 de ≈ 230 ms à ≈ 1 500 ms
50	TP 10 : lignes ignorées par le watermark
8h27–8h29	TP 10 : fenêtres absentes du Parquet, pas encore closes
UTC T13:27:55Z	TP 10 : le watermark de lastProgress = 8h27 à New York
30 min, 10 min	TP 11 : fenêtre et watermark
0,21	TP 11 : écart moyen prédit / réel (fenêtres de 30 à 70 courses)
≈ 0,4 s / ≈ 0,15 s	TP 11 : CSV / Parquet, × 2 à 3
3 colonnes sur 19	TP 11 : ReadSchema, en CSV comme en Parquet
2	TP 11 : blocs de whole-stage codegen
250 / 600–1 000 lignes/s	Spark UI : Input Rate / Process Rate, le flux tient
3.0, 4.0, 4.1	Flume retiré / SparkR déprécié et Mesos retiré / real-time mode

Questions que pose un formateur

1. Pourquoi un stream exige-t-il un schéma explicite ? Spark ne peut pas inférer un schéma sur des fichiers qui n’existent pas encore. On le reprend sur trips.parquet.
2. Vous relancez le TP 9 après un --reset et rien ne s’affiche. Pourquoi ? Le checkpoint a gardé l’ancien état et les fichiers déjà lus. Il faut le supprimer avant de relancer.
3. Le stream compte moins de courses que le batch sur les mêmes fichiers. Bug ? Non, un retard : 2 fichiers par micro-batch de 5 s contre 1 fichier toutes les 2 s. Une fois le simulateur arrêté, le stream rattrape.
4. Pourquoi les dernières minutes n’apparaissent-elles pas dans le Parquet du TP 10 ? En append, une fenêtre n’est écrite qu’une fois close par le watermark. Les dernières n’ont pas encore dépassé max(ts) − 2 minutes.
5. Quelle différence entre complete, update et append ? complete réécrit toute la table, update émet les lignes modifiées, append n’émet une ligne qu’une fois, quand elle est définitive.
6. Pourquoi faire la moyenne de p_pourboire plutôt que des prédictions 0/1 ? La moyenne des probabilités estime directement un taux comparable au taux réel. Les 0/1 dépendent du seuil.
7. Les deux plans, CSV et Parquet, montrent PushedFilters. D’où vient l’écart de vitesse ? Du format. Le CSV lit et découpe chaque ligne. Parquet saute les colonnes et les blocs (min/max), puis lit en vectorisé.
8. Lambda ou Kappa pour un tableau de bord, des alertes, une facturation ? Le flux suffit pour le tableau de bord et les alertes (Kappa). La facturation exacte justifie un recalcul batch : architecture hybride.
9. Que reste-t-il de Flume et des DStreams ? À savoir lire dans l’existant. Kafka a remplacé Flume (connecteur retiré en 3.0), Structured Streaming a remplacé les DStreams (legacy).