TP 10 – Les logs en temps réel

• 35 min · 11h10 – 11h45
• Notebook : jour3/TP10_logs_temps_reel.ipynb · corrigé Scala : scala/tp10_logs_temps_reel.scala
• Simulateur : scripts/stream_simulator.py logs

📁 Dans le dépôt

• Notebook : jour3/TP10_logs_temps_reel.ipynb
• Données : scripts/stream_simulator.py (stream/logs/*.csv)
• Corrigé : jour3/corriges/tp10_logs_temps_reel.py (publié en fin de J3)

À exécuter depuis la racine du dépôt. Pas encore cloné ? Voir Prérequis et installation.

L’application de réservation écrit ses logs dans stream/logs/. Une panne survient quelque part dans la matinée.

Objectif : la détecter en temps réel.

Réussi si : une alerte (taux d'erreur > 10 %) apparaît sur le bon endpoint, aux bonnes minutes.

Avant de commencer

python scripts/stream_simulator.py logs --reset          # en direct : un fichier toutes les 2 secondes
# ou, si vous êtes en retard / en mode dégradé :
python scripts/stream_simulator.py logs --all --reset    # tout d'un coup

Exemple de ligne de log :

2024-01-15T08:22:03.120 ERROR endpoint=/payment status=503 latency_ms=812 zone=161

Étapes

1. Lire les logs en continu

readStream.text("stream/logs") : une colonne value par ligne. Avec .option("maxFilesPerTrigger", 3) pour avoir plusieurs micro-batches, même avec --all.

2. Parser avec regexp_extract

Motif fourni :

motif = r"^(\S+) +(\w+) +endpoint=(\S+) status=(\d+) latency_ms=(\d+) zone=(\d+)$"

Colonnes : ts (to_timestamp(..., "yyyy-MM-dd'T'HH:mm:ss.SSS")), niveau, endpoint, status (int), latence_ms (int).

3. Watermark de 2 minutes, fenêtre de 1 minute par endpoint

Dans agg : requetes (count), erreurs (sum(when(status >= 500, 1).otherwise(0))), latence_p95 (percentile_approx(latence_ms, 0.95)) ; puis taux_erreur = erreurs / requetes avec withColumn.

4. Écrire en mode append vers Parquet (out/tp10/par_minute), avec un checkpoint

En direct : trigger(processingTime="5 seconds") et awaitTermination(120). Avec --all : trigger(availableNow=True). Supprimez out/checkpoints/tp10 et out/tp10 avant de démarrer.

5. Lire le Parquet (en batch) et afficher les alertes : taux d’erreur > 10 %

Quel endpoint ? À quelles minutes ? Que devient la latence p95 ?

6. Combien de lignes ont été ignorées par le watermark ?

Indice : q.recentProgress, champ stateOperators[0]["numRowsDroppedByWatermark"]. Pourquoi les toutes dernières minutes n’apparaissent-elles pas dans le Parquet ?

Note

lastProgress affiche le watermark en UTC (…T13:27:55Z pour 8h27 à New York). Une fenêtre de 7h52 peut apparaître : ce sont des lignes arrivées avec 8 minutes de retard au tout début du flux.

Bonus – Fenêtre glissante de 5 minutes, toutes les minutes

L’alerte est-elle plus stable ?

Correction

import shutil
from pyspark.sql import functions as F
spark.conf.set("spark.sql.shuffle.partitions", "4")
for d in ["out/checkpoints/tp10", "out/tp10"]:
    shutil.rmtree(d, ignore_errors=True)

# 1. Une ligne de texte par événement
brut = spark.readStream.option("maxFilesPerTrigger", 3).text("stream/logs")

# 2. Parser
motif = r"^(\S+) +(\w+) +endpoint=(\S+) status=(\d+) latency_ms=(\d+) zone=(\d+)$"
logs = brut.select(
    F.to_timestamp(F.regexp_extract("value", motif, 1), "yyyy-MM-dd'T'HH:mm:ss.SSS").alias("ts"),
    F.regexp_extract("value", motif, 2).alias("niveau"),
    F.regexp_extract("value", motif, 3).alias("endpoint"),
    F.regexp_extract("value", motif, 4).cast("int").alias("status"),
    F.regexp_extract("value", motif, 5).cast("int").alias("latence_ms"),
)

# 3. Fenêtre de 1 minute sur le temps de l'ÉVÉNEMENT, watermark de 2 minutes
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"),
                   F.round(F.expr("percentile_approx(latence_ms, 0.95)"), 0).alias("latence_p95"))
              .withColumn("taux_erreur", F.round(F.col("erreurs") / F.col("requetes"), 3)))

# 4. append vers Parquet : une fenêtre n'est écrite qu'UNE fois, quand le watermark l'a close
q = (par_minute.writeStream
     .outputMode("append")
     .format("parquet")
     .option("path", "out/tp10/par_minute")
     .option("checkpointLocation", "out/checkpoints/tp10")
     .trigger(processingTime="5 seconds")        # ou trigger(availableNow=True) avec --all
     .start())
q.awaitTermination(120)
q.stop()

# 5. Lire en batch et détecter l'incident
res = spark.read.parquet("out/tp10/par_minute")
(res.filter("endpoint = '/payment'")
    .select(F.date_format("window.start", "HH:mm").alias("minute"), "requetes", "erreurs", "taux_erreur", "latence_p95")
    .orderBy("minute").show(40))
res.filter("taux_erreur > 0.10") \
   .select(F.date_format("window.start", "HH:mm").alias("minute"), "endpoint", "taux_erreur").orderBy("minute").show()

# 6. Lignes ignorées par le watermark
micro_batchs = q.recentProgress                    # un dictionnaire de métriques par micro-batch
ignorees = sum(int(mb["stateOperators"][0].get("numRowsDroppedByWatermark", 0))
               for mb in micro_batchs if mb["stateOperators"])
print("Lignes ignorées car trop en retard (> 2 min) :", ignorees)

Résultats attendus

Mesure	Valeur
Endpoint en panne	/payment
Minutes en alerte	8h22 à 8h25
Taux d’erreur de /payment	de ≈ 1 % à 29–38 %
Latence p95	de ≈ 230 ms à ≈ 1 500 ms
Lignes ignorées par le watermark	50
Fenêtres absentes du Parquet	8h27–8h29 : pas encore closes en append

# Bonus – fenêtres glissantes : 5 minutes, toutes les minutes (chaque événement compte dans 5 fenêtres)
glissant = (spark.read.text("stream/logs")
            .select(F.to_timestamp(F.regexp_extract("value", motif, 1), "yyyy-MM-dd'T'HH:mm:ss.SSS").alias("ts"),
                    F.regexp_extract("value", motif, 4).cast("int").alias("status"))
            .groupBy(F.window("ts", "5 minutes", "1 minute"))
            .agg(F.round(F.avg((F.col("status") >= 500).cast("int")), 3).alias("taux_erreur_5min"))
            .orderBy("window"))

La fenêtre glissante lisse le signal : l’alerte monte plus progressivement, mais résiste mieux aux pics isolés.

🔍 Décryptage de la regex : Comment regexp_extract découpe-t-elle la ligne de log ?

Reprenons la ligne brute reçue dans la colonne value : 2024-01-15T08:22:03.120 ERROR endpoint=/payment status=503 latency_ms=812 zone=161

Et notre expression régulière contenant 6 groupes entre parenthèses (…) : r"^(\S+) +(\w+) +endpoint=(\S+) status=(\d+) latency_ms=(\d+) zone=(\d+)$"

Numéro de groupe (idx dans regexp_extract)	Motif	Ce qu’il capture sur l’exemple
Groupe 1	(\S+) (caractères non-espaces)	"2024-01-15T08:22:03.120" ➔ converti en ts
Groupe 2	(\w+) (mot)	"ERROR" ➔ colonne niveau
Groupe 3	endpoint=(\S+)	"/payment" ➔ colonne endpoint
Groupe 4	status=(\d+) (chiffres)	"503" ➔ converti en entier status
Groupe 5	latency_ms=(\d+)	"812" ➔ converti en entier latence_ms
Groupe 6	zone=(\d+)	"161" ➔ identifiant de zone

Pièges rencontrés en séance

Symptôme	Cause	Solution
Append output mode not supported when there are streaming aggregations without watermark	Watermark oublié	withWatermark("ts", "2 minutes") avant le groupBy
ts est null partout	Format de date	yyyy-MM-dd'T'HH:mm:ss.SSS (le T entre quotes)
Rien dans le Parquet	Aucune fenêtre encore close	Attendre les micro-batches suivants ; vérifier numRowsDroppedByWatermark
Watermark affiché à 13h27	UTC	8h27 à New York
Fenêtre de 7h52	Lignes très en retard au début du flux	Normal

Partage

• Un participant présente son alerte et sa courbe.
• Pourquoi 8h27–8h29 n’apparaissent-elles pas ? Fenêtres pas encore closes.
• Watermark de 2 min ou de 10 min pour une alerte ? Court pour la réactivité ; long ou batch de rattrapage pour l’exhaustivité.
• À retenir : le watermark arbitre entre réactivité et exhaustivité.