Partie 1 – Structured Streaming : traiter les données qui arrivent

• 9h20 – 10h30
• Théorie 35 min · TP 9 35 min
• Slides 6 à 13

Et si votre tableau de bord se mettait à jour toutes les 5 secondes, sans relancer aucun job ? Hier soir, combien de temps aurait-il fallu pour savoir qu'une panne avait lieu ? Avec un batch toutes les nuits : le lendemain.

Un flux = une table qui grandit sans fin

 t = 1                    t = 2                    t = 3
 ┌────────────┐           ┌────────────┐           ┌────────────┐
 │ lignes     │           │ lignes     │           │ lignes     │
 │            │           │ + nouvelles│           │ + nouvelles│
 └────────────┘           │            │           │ + nouvelles│
   même requête           └────────────┘           └────────────┘
   Manhattan : 2 400        même requête             même requête
                            Manhattan : 4 850        Manhattan : 7 300

La table d’entrée est « non bornée » : chaque micro-batch ajoute des lignes, et Spark met le résultat à jour de façon incrémentale, sans tout recalculer. Il garde un état : ici, le total courant par quartier. Analogie : un compteur de supporters au stade. On n’en recompte pas 40 000 à chaque entrée, on ajoute ceux qui passent le portique.

	DataFrame statique	DataFrame dynamique
Création	spark.read	spark.readStream
Nature	Borné, calculé une fois	Non borné, calculé en continu
Test	df.isStreaming == False	df.isStreaming == True

C’est le modèle mental de Structured Streaming : la même API DataFrame s’applique aux jeux de données statiques (fichiers déjà présents) et dynamiques (fichiers qui apparaissent au fil du temps).

Le code : presque comme en batch

Le même calcul (courses par arrondissement) écrit deux fois. Dans les deux cas, schema est pris sur les données statiques et zones est le référentiel data/taxi_zone_lookup.csv, lu comme un DataFrame ordinaire : le JSON du flux ne contient que PULocationID, l’arrondissement vient de la jointure.

# Batch, comme au jour 1 (bonus du corrigé TP 9)
schema = spark.read.parquet("data/trips.parquet").schema
zones = spark.read.csv("data/taxi_zone_lookup.csv", header=True, inferSchema=True)
ref = zones.select(F.col("LocationID").alias("PULocationID"), "Borough")

df = spark.read.schema(schema).json("stream/courses")
res = df.join(ref, "PULocationID").groupBy("Borough").count()
res.write.mode("overwrite").parquet("out/res")

# Streaming, jour 3 (TP 9) : seules les lignes marquées ← changent
df = spark.readStream.schema(schema).json("stream/courses")          # ← readStream, schéma obligatoire
res = df.join(ref, "PULocationID").groupBy("Borough").count()       #   identique

q = (res.writeStream                                                  # ← writeStream
     .outputMode("complete")                                          # ← quoi écrire à chaque micro-batch
     .format("console")
     .option("checkpointLocation", "out/checkpoints/tp9")             # ← la mémoire de la requête
     .trigger(processingTime="5 seconds")                             # ← quand déclencher
     .start())

Quatre différences, pas plus :

Différence	Pourquoi
readStream / writeStream	Le reste de l’API ne change pas
Schéma obligatoire	Le flux est infini : Spark ne peut pas le lire en entier pour deviner les types
outputMode	Quoi écrire à chaque micro-batch (détaillé en partie 2)
trigger	Quand déclencher un micro-batch

Trigger	Comportement	Usage
processingTime="5 seconds"	Un micro-batch toutes les 5 s	Nos TP en direct
availableNow=True	Traite tout ce qui est disponible, puis s’arrête	Jobs planifiés, nos corrigés reproductibles
realTime="5 minutes" (mode temps réel, Spark 4.1 ; mot-clé PySpark depuis 4.2, Trigger.RealTime en Scala/Java avant)	Traitement continu, latence de quelques millisecondes ; la durée est celle d’un run avant redémarrage	Sources Kafka, mode update, opérations sans état au départ

start() rend la main tout de suite : la requête tourne en arrière-plan. q.awaitTermination(), q.stop(), q.lastProgress, q.recentProgress la pilotent. Le checkpoint est obligatoire dès qu’on sort de la console : c’est la mémoire de la requête (partie 2).

Sources et sinks

Sources : d’où vient le flux	Sinks : où vont les résultats
Fichiers (CSV, JSON, Parquet…) : un dossier surveillé. Nos TP	Fichiers (Parquet, Delta…) et Kafka
Kafka : la référence en production, rejouable	foreachBatch : du code batch libre pour chaque micro-batch
Rate et socket : pour tester uniquement	Console et memory : pour le développement
Une source doit pouvoir être rejouée pour la tolérance aux pannes	Le sink décide de la garantie : exactly-once ou at-least-once

Bonne pratique pour une source fichiers

Écrire le fichier ailleurs puis le déplacer (rename atomique), sinon Spark peut lire un fichier à moitié écrit. C’est ce que fait notre simulateur.

Kafka est un journal distribué, partitionné, rejouable ; le connecteur spark-sql-kafka est un package à ajouter (--packages). foreachBatch sert à écrire dans plusieurs destinations, faire un MERGE dans une table ou appeler une API : on l’utilise au TP 11.

L’héritage Flume : agents, sources, channels, sinks, Avro RPC

 logs des serveurs ──► ┌─ Agent Flume 1 ─────────┐   Avro RPC   ┌─ Agent Flume 2 ─────────┐ ──► HDFS
                       │ Source → Channel → Sink │ ───────────► │ Source → Channel → Sink │
                       └─────────────────────────┘              └─────────────────────────┘

Brique Flume	Rôle	Équivalent moderne
Source	Reçoit les événements (logs, syslog, HTTP)	Producteur Kafka
Channel	Tampon entre source et sink (mémoire ou fichier)	Topic Kafka
Sink	Écrit vers la destination (HDFS, un autre agent)	Consommateur Kafka
Avro RPC	Sérialisation binaire compacte, avec schéma, entre agents	Avro reste très utilisé : schémas Avro dans Kafka, from_avro / to_avro dans Spark

Pourquoi nous l’enseignons

Apache Flume collecte et achemine des logs via des agents composés de sources, channels et sinks, avec une sérialisation Avro RPC entre agents. Nous le présentons parce que vous le rencontrerez dans des pipelines historiques — mais le connecteur spark-streaming-flume a été retiré en Spark 3.0. Pour un nouveau projet, utilisez Kafka ; Avro reste pertinent au niveau du format (schémas Avro dans Kafka, from_avro / to_avro).

StreamingContext ou Structured Streaming ?

	Spark Streaming (DStreams)	Structured Streaming
Point d’entrée	StreamingContext(sc, 5)	SparkSession, API DataFrame et SQL
Modèle	Un flux = une suite de RDD (un par micro-batch)	Une table non bornée, optimisée par Catalyst comme le batch
Temps	Temps de traitement seulement	Temps d’événement, watermark, exactly-once
Statut	Legacy : plus aucune mise à jour	Spark 4.1 : real-time mode, latence en millisecondes

# À savoir lire dans du code existant, pas à écrire
ssc = StreamingContext(sc, 5)
lignes = ssc.textFileStream("dir")
lignes.count().pprint()
ssc.start()

Pourquoi nous l’enseignons

StreamingContext est le point d’entrée des DStreams, la génération précédente de l’API de streaming. La documentation officielle précise aujourd’hui : « Spark Streaming is the previous generation of Spark’s streaming engine. There are no longer updates to Spark Streaming and it’s a legacy project. » Savoir le lire dans l’existant suffit ; tout nouveau projet s’écrit en Structured Streaming.

À vous de jouer

TP 9 – Votre premier stream (35 min) : le chiffre d’affaires par quartier, en direct, mis à jour à chaque micro-batch.

Pause de 15 minutes ensuite : arrêtez les requêtes (q.stop()) et le simulateur (Ctrl+C). Puis la partie 2 : fenêtres et watermark.