TP 9 – Votre premier stream

• 35 min · 9h55 – 10h30
• Notebook : jour3/TP9_premier_stream.ipynb
• Simulateur : scripts/stream_simulator.py courses

📁 Dans le dépôt

• Notebook : jour3/TP9_premier_stream.ipynb
• Données : scripts/stream_simulator.py (dépose stream/courses/*.json)
• Corrigé : jour3/corriges/tp9_premier_stream.py (publié en fin de J3)

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

Objectif : suivre en direct le chiffre d’affaires et le nombre de courses par arrondissement.

Réussi si : le tableau se met à jour à chaque micro-batch.

Avant de commencer

Dans un terminal, à la racine du dépôt :

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

Les fichiers apparaissent dans stream/courses : c’est notre « production ». Dans le notebook :

from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.appName("TP").master("local[*]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
spark.conf.set("spark.sql.shuffle.partitions", "4")   # petit volume : 200 partitions par défaut, c'est trop

Étapes

1. Lire le dossier stream/courses en continu

Un stream exige un schéma explicite : prenez celui de data/trips.parquet (spark.read.parquet(...).schema). Indices : spark.readStream.schema(schema).option("maxFilesPerTrigger", 2).json("stream/courses"), puis isStreaming.

2. Jointure stream-statique avec les zones

Le référentiel data/taxi_zone_lookup.csv est un DataFrame statique ordinaire. Ajoutez la colonne Borough.

3. Chiffre d’affaires et nombre de courses par Borough

groupBy("Borough").agg(count, sum(total_amount)).

4. Démarrer la requête

outputMode("complete"), format("console"), trigger(processingTime="5 seconds"), un checkpointLocation. Laissez tourner 60 secondes : q.awaitTermination(60), puis q.stop(). Les tableaux s’affichent sous la cellule, un par micro-batch.

Le checkpoint est la mémoire de la requête

Si vous relancez après un --reset du simulateur, supprimez d’abord out/checkpoints/tp9, sinon la requête reprend l’ancien état et ignore les nouveaux fichiers.

5. Le suivi : q.lastProgress, q.recentProgress

Combien de micro-batches ? Combien de lignes par micro-batch (numInputRows) ?

Bonus – Le même calcul en batch (spark.read) sur les mêmes fichiers

Obtenez-vous les mêmes comptes ? Si le stream en a moins : comparez le rythme du simulateur (1 fichier toutes les 2 s) et maxFilesPerTrigger (2 fichiers par micro-batch de 5 s). Le stream a-t-il pris du retard ? Que se passe-t-il si on le laisse tourner plus longtemps ?

Correction

import shutil
shutil.rmtree("out/checkpoints/tp9", ignore_errors=True)   # repartir de zéro

# 1. 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 : on voit le flux avancer
           .json("stream/courses"))
print("isStreaming :", courses.isStreaming)            # True

# 2. Jointure stream-statique
zones = spark.read.csv("data/taxi_zone_lookup.csv", header=True, inferSchema=True)
enrichies = courses.join(zones.select(F.col("LocationID").alias("PULocationID"), "Borough"), "PULocationID")

# 3. Agrégation continue depuis le début du flux
par_borough = (enrichies.groupBy("Borough")
               .agg(F.count("*").alias("courses"), F.round(F.sum("total_amount"), 2).alias("ca")))

# 4. Démarrer : complete = toute la table résultat à chaque micro-batch
q = (par_borough.writeStream
     .outputMode("complete")
     .format("console").option("truncate", False)
     .trigger(processingTime="5 seconds")
     .option("checkpointLocation", "out/checkpoints/tp9")
     .start())
q.awaitTermination(60)
q.stop()

# 5. Suivi
batches = q.recentProgress
print(f"{len(batches)} micro-batches, {sum(p['numInputRows'] for p in batches)} courses traitées")
print({k: batches[-1][k] for k in ("batchId", "numInputRows")})

# Bonus – le même calcul en batch
(spark.read.schema(schema).json("stream/courses")
      .join(zones.select(F.col("LocationID").alias("PULocationID"), "Borough"), "PULocationID")
      .groupBy("Borough").agg(F.count("*").alias("courses")).orderBy(F.desc("courses")).show())

Résultats attendus

Avec le flux complet : Manhattan 11 880 courses, Queens 1 553… soit 14 992 courses en 15 micro-batches. Le batch sur les mêmes fichiers donne exactement les mêmes comptes : le streaming est un batch incrémental.

Pièges rencontrés en séance

Symptôme	Cause	Solution
Schema must be specified when creating a streaming source DataFrame	Schéma oublié	.schema(schema) repris de trips.parquet
Rien ne s’affiche après un --reset	Le checkpoint a gardé l’ancien état et les fichiers déjà lus	Supprimer out/checkpoints/tp9 avant de relancer
Micro-batches lents (plusieurs secondes)	spark.sql.shuffle.partitions = 200	Le passer à 4
Le stream compte moins que le batch	2 fichiers par micro-batch de 5 s contre 1 fichier toutes les 2 s : il prend du retard	Le laisser tourner : une fois le simulateur arrêté, il rattrape
Requête modifiée, erreur au redémarrage	Même checkpoint pour une requête différente	Nouveau checkpoint

Partage

• Le batch donne-t-il le même résultat ? Oui : le streaming est un batch incrémental.
• Pourquoi un schéma explicite ? Spark ne peut pas inférer un schéma sur des fichiers qui n’existent pas encore.
• À retenir : readStream / writeStream, un schéma, un output mode, un trigger, un checkpoint. Le reste est du batch.