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.

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.

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.