Partie 2 – Agrégations continues, fenêtres et analyse de logs

• 10h45 – 12h00
• Théorie 25 min · TP 10 35 min · partage et quiz flash 15 min
• Slides 14 à 23

Et si une panne du paiement apparaissait à 8h22 : combien de minutes pour la voir ?

Fenêtres : regrouper le temps

 temps de l'événement   8h00    8h01    8h02    8h03    8h04    8h05
                        │       │       │       │       │       │
 Fixes (tumbling), 1 min  [8h00–8h01][8h01–8h02][8h02–8h03][8h03–8h04][8h04–8h05]

 Glissantes (sliding),    [8h00 ──────── 8h03]
 3 min toutes les 1 min           [8h01 ──────── 8h04]
                                          [8h02 ──────── 8h05]

df.groupBy(F.window("ts", "1 minute"))                 # fixe
df.groupBy(F.window("ts", "3 minutes", "1 minute"))    # glissante
df.groupBy(F.session_window("ts", "5 minutes"))        # session

Fenêtre	Chaque événement compte dans…	Usage
Fixe	1 fenêtre	Tableaux de bord (requêtes par minute)
Glissante	Plusieurs fenêtres → signal lissé	Alertes plus stables (taux d’erreur sur 5 min, recalculé chaque minute)
Session	La session de l’utilisateur, close après 5 min d’inactivité	Parcours d’un utilisateur

Une fenêtre est simplement une colonne de groupement calculée à partir du temps de l’événement (ts), pas de l’heure d’arrivée. Question au groupe : avec une fenêtre de 3 min glissant toutes les minutes, dans combien de fenêtres compte un événement ? Trois.

Le watermark : jusqu’où attendre les retardataires ?

 temps de l'événement  8h00   8h02   8h04   8h06   8h08   8h10
                                                   │      │
                                           watermark    max vu
                                            = 8h08      = 8h10      tolérance : 2 min
   arrive en retard, ts = 8h09 ──► acceptée (≥ watermark)
   arrive en retard, ts = 8h05 ──► ✕ ignorée (< watermark)

logs.withWatermark("ts", "2 minutes").groupBy(F.window("ts", "1 minute"), "endpoint").count()

• Watermark = plus grand temps d’événement vu − tolérance. Plus ancien que le watermark : ignoré.
• Il permet de clore les fenêtres et de libérer la mémoire (l’état). Sans lui, l’état grossit à l’infini.
• Le compromis : tolérance longue = plus juste, mais plus lent et plus de mémoire ; courte = réactif, mais perd des retardataires.
• Au TP 10 : 50 lignes de log arrivées avec plus de 2 min de retard ont été ignorées (numRowsDroppedByWatermark dans q.lastProgress).

Temps d’événement (quand c’est arrivé) ≠ temps de traitement (quand Spark le reçoit) : un téléphone hors ligne, un réseau lent, un serveur qui redémarre… les données arrivent en retard et dans le désordre. C’est la notion la plus importante du streaming, et la plus contre-intuitive.

🔍 Vulgarisation : Temps d'événement vs Temps de traitement (l'analogie du tunnel sous l'Hudson)

Imaginez qu’un taxi termine sa course à 8h04 sous le tunnel Lincoln (pas de réseau 4G/5G). Le compteur enregistre dans le JSON : "tpep_pickup_datetime": "08:04:12" (Temps d’événement / Event Time). Sept minutes plus tard, à 8h11, le taxi sort du tunnel, retrouve la 4G et envoie enfin l’événement à votre serveur Spark (Temps de traitement / Processing Time = 8h11).

• Si vous groupiez par heure d’arrivée sur le serveur (8h11) : votre graphique dirait qu’il y a eu un embouteillage à 8h11 alors qu’il a eu lieu à 8h04 !
• Avec F.window("ts", "1 minute"), Spark regarde l’horodatage 08:04:12 écrit à l’intérieur de l’événement et range la course dans la boîte [08:04 – 08:05], même si elle est arrivée en retard.

🧮 Trace minute par minute : Comment le Watermark ferme les fenêtres et déclenche l'écriture dans Parquet (mode Append)

Suivons une requête avec F.window("ts", "1 minute") et withWatermark("ts", "2 minutes") en mode append (qui n’écrit une fenêtre dans Parquet que lorsqu’il est certain à 100 % qu’elle ne changera plus jamais) :

Micro-batch	Événement reçu (ts)	Plus grand ts vu (max_ts)	Seuil Watermark (max_ts - 2 min)	Que fait Spark sur les fenêtres en mémoire ?
Batch 1	08:01:10, 08:02:05	08:02:05	08:00:05	Les fenêtres [08:01-08:02] et [08:02-08:03] sont ouvertes en RAM. 0 fichier Parquet écrit (on attend d’éventuels retardataires).
Batch 2	08:03:40	08:03:40	08:01:40	Le watermark (08:01:40) reste inférieur à la fin de [08:01-08:02] (08:02:00). Toujours 0 écriture Parquet.
Batch 3	08:04:15 + un retardataire de 08:01:50	08:04:15	08:02:15	1) Le retardataire 08:01:50 est accepté. 2) Le nouveau watermark (08:02:15) dépasse enfin la fin de la fenêtre [08:01-08:02] (08:02:00) ! Spark clôt la fenêtre [08:01-08:02], l’écrit dans Parquet et libère sa mémoire.
Batch 4	Un gros retardataire de 08:01:20 arrive	08:04:15	08:02:15	08:01:20 < 08:02:15 (plus vieux que le watermark, sa fenêtre est déjà close et écrite) ➔ ignoré (numRowsDroppedByWatermark + 1).

C’est exactement pour cela qu’au TP 10, lorsque le flux s’arrête à 08:29, le watermark vaut 08:27 : les fenêtres 08:27–08:28 et 08:28–08:29 restent ouvertes en mémoire et n’apparaissent pas encore dans le dossier Parquet !

Output modes et tolérance aux pannes

Output mode	Ce qui est écrit à chaque micro-batch	Usage
append	Seulement les lignes définitives (fenêtre close)	Écrire des fichiers : chaque fenêtre une seule fois
update	Les lignes qui ont changé	Tableau de bord, base clé-valeur
complete	Toute la table résultat	Petits résultats (console, memory)

Agrégation + append = watermark obligatoire

Sans watermark, Spark refuse : Append output mode not supported when there are streaming aggregations without watermark. Conséquence visible au TP 10 : en append, les 2 ou 3 dernières minutes n’apparaissent pas encore dans le Parquet, elles ne sont pas closes.

Checkpoint : la mémoire de la requête.

• Enregistre les offsets lus et l’état (totaux, fenêtres). Dossier HDFS, S3 ou disque, obligatoire hors console.
• Après une panne : reprise là où on s’était arrêté.
• Source rejouable + sink idempotent = exactly-once : chaque événement compté une fois exactement, même après une panne (fichiers, Kafka en source ; fichiers, Delta, foreachBatch idempotent en sink).
• Changer la requête ? Nouveau checkpoint.

Spark 4.0 ajoute le State Data Source pour lire l’état d’une requête, et transformWithState pour les traitements avec état sur mesure.

🔍 Comparatif visuel : Append vs Update vs Complete sur un exemple simple

Supposons que l’on compte les erreurs par minute :

• Au Batch 1, on a : [08:00-08:01] → 5 erreurs, [08:01-08:02] → 2 erreurs.
• Au Batch 2, l’horloge avance (le watermark clôt 08:00-08:01), 08:01-08:02 passe de 2 à 3 erreurs, et 08:02-08:03 apparaît avec 1 erreur.

Que sort Spark au Batch 2 selon l’outputMode choisi ?

• complete (console / petits tableaux) : réécrit les 3 lignes (08:00→5, 08:01→3, 08:02→1), même celles qui n’ont pas bougé.
• update (tableaux de bord temps réel / Redis / PostgreSQL) : n’émet que les 2 lignes qui ont changé ce tour-ci (08:01→3 et 08:02→1).
• append (fichiers Parquet / Delta Lake) : n’émet que la ligne définitivement close par le watermark (08:00→5), car on ne peut pas modifier un fichier Parquet déjà écrit sur disque !

Encoders et Datasets typés

// Scala
case class Course(PULocationID: Int, trip_distance: Double, total_amount: Double)
val ds: Dataset[Course] = df.as[Course]      // Encoder implicite
ds.filter(c => c.trip_distance > 10)          // vérifié à la compilation
  .map(c => c.total_amount / c.trip_distance)

• Un Encoder traduit un objet JVM (Course) en format binaire Tungsten, et inversement.
• Dataset[T] : typé, erreurs détectées à la compilation (Scala, Java). DataFrame = Dataset[Row] : non typé, le même moteur. Un flux peut aussi être un Dataset typé : spark.readStream…as[Course].
• Python n’a pas de Dataset typé : DataFrame + schéma.

Performance

Les lambdas typées sont opaques pour Catalyst : ds.filter(c => c.trip_distance > 10) oblige Spark à reconstruire chaque objet pour appeler la lambda. df.filter(F.col("trip_distance") > 10) est compris par Catalyst, poussé vers la lecture et compilé par Tungsten. Préférer les colonnes. On le voit en partie 3.

À vous de jouer

TP 10 – Les logs en temps réel (35 min) : une vraie panne vous attend dans les logs. Réussi si l’alerte sur /payment apparaît entre 8h22 et 8h25.

Partage et remédiation

• Pourquoi 8h27–8h29 n’apparaissent-elles pas dans le Parquet ? Les dernières fenêtres ne sont pas closes : le watermark ne les a pas encore dépassées. Elles seront écrites quand de nouvelles données arriveront.
• Watermark de 2 min ou de 10 min : lequel pour une alerte ? Court pour la réactivité. Pour un rapport de facturation : long, ou un batch de rattrapage, pour l’exhaustivité. C’est le début de la discussion Lambda / Kappa de cet après-midi.
• À retenir : le watermark arbitre entre réactivité et exhaustivité.

Quiz flash : 4 situations de production

Vote à main levée ou Slido, puis une minute de débat par question. Réponses dans le quiz du jour 3.

1. Mon agrégation en mode append n’écrit rien dans le Parquet. Pourquoi ?
2. Ma requête redémarre après une panne. Recommence-t-elle depuis le début ?
3. Une ligne arrive avec 10 min de retard, watermark de 2 min. Que devient-elle ?
4. count(distinct user) sur un flux : possible ?

Avant le déjeuner : l’émargement, et on arrête les simulateurs. Reprise à 13h00 avec la partie 3.