Quiz 09 – Structured Streaming : premiers flux et analyse de logs

• Jour 3
• 20 questions
• Seuil de réussite : 70 %
• 20 flashcards

Contenus évalués : lecture d'un dossier en continu avec `readStream` et schéma explicite ; écriture avec `writeStream`, `trigger` (`processingTime`, `availableNow`) et `checkpointLocation` ; suivi avec `recentProgress` et `lastProgress` ; reprise après panne et rôle de `--reset` ; jointure flux-statique avec le référentiel des zones ; output modes `append`, `update` et `complete` ; fenêtres temporelles fixes et glissantes sur le temps de l'événement ; watermark, lignes en retard ignorées et compromis réactivité-exhaustivité ; opérations non supportées sur un flux (distinct exact, tri sans agrégation, append sans watermark) ; `stream_simulator.py` et l'arrivée des données fichier par fichier.

Une seule bonne réponse par question. Toutes les valeurs citées sont celles des TP de la formation. Le corrigé détaillé s’affiche après la correction. Ce quiz prépare le QCM de validation des acquis du dernier jour.

QCM

Question 1

Dans le TP 9, tu exécutes ce code. Qu'affiche le print ?

schema = spark.read.parquet("data/trips.parquet").schema
courses = (spark.readStream
           .schema(schema)
           .option("maxFilesPerTrigger", 2)
           .json("stream/courses"))
print(courses.isStreaming)

• a) False, car aucune action n'a encore été lancée sur courses
• b) True, car courses est un DataFrame dynamique alimenté en continu
• c) Le nombre de fichiers déjà présents dans stream/courses
• d) Une erreur, car un flux ne peut pas lire du JSON avec le schéma d'un fichier Parquet

Question 2

Le simulateur stream_simulator.py écrit chaque fichier dans un sous-dossier _tmp, puis le déplace avec os.replace dans stream/courses ou stream/logs. Pourquoi ?

• a) Pour compresser les fichiers avant que Spark ne les lise
• b) Pour que Spark relise les anciens fichiers à chaque micro-batch
• c) Pour que Spark ne voie jamais un fichier à moitié écrit
• d) Pour imposer l'ordre des événements à l'intérieur de chaque fichier

Question 3

Dans le TP 9, zones est lu avec spark.read.csv("data/taxi_zone_lookup.csv", ...) et joint au flux courses pour ajouter la colonne Borough. Quelle affirmation est juste ?

• a) zones reste un DataFrame statique ordinaire, et chaque micro-batch de courses est joint à ce référentiel
• b) Il faut lire zones avec spark.readStream, sinon la jointure est refusée
• c) La jointure n'est possible qu'en mode append, donc pas avec l'agrégation par Borough
• d) Spark recharge zones comme un nouveau flux à chaque fichier de courses

Question 4

Le TP 9 démarre la requête par_borough avec outputMode("complete"). Que reçoit la console à chaque micro-batch ?

• a) Uniquement les nouvelles lignes arrivées depuis le micro-batch précédent
• b) Uniquement les arrondissements dont les comptes ont changé
• c) Uniquement les fenêtres closes par le watermark
• d) Toute la table résultat, chiffre d'affaires et nombre de courses par Borough depuis le début

Question 5

La requête du TP 10 s'arrête brutalement en pleine matinée. Tu la relances avec le même checkpointLocation (out/checkpoints/tp10), sans rien supprimer. Que se passe-t-il ?

• a) Elle relit tous les fichiers de stream/logs depuis le début et recalcule tout
• b) Elle reprend aux derniers offsets enregistrés, avec son état
• c) Elle ignore le checkpoint, car il ne sert qu'au mode complete
• d) Elle refuse de démarrer tant que le dossier out/tp10 n'est pas vidé

Question 6

Dans le TP 10, le watermark est de 2 minutes sur ts. En milieu de matinée, alors que le watermark a déjà avancé, une ligne de log arrive avec 5 minutes de retard, comme le simulateur en produit. Que devient-elle ?

• a) Elle est comptée dans la fenêtre de la minute où elle est arrivée
• b) Elle rouvre sa fenêtre, qui est réécrite dans le Parquet
• c) Elle est ignorée et comptée dans numRowsDroppedByWatermark
• d) Elle est mise en attente jusqu'au prochain redémarrage de la requête

Question 7

À la fin du TP 10, tu lis out/tp10/par_minute en batch. Les fenêtres de 8h27 à 8h29 n'y figurent pas. La requête a écrit avec ce code :

q = (par_minute.writeStream
     .outputMode("append")
     .format("parquet")
     .option("path", "out/tp10/par_minute")
     .option("checkpointLocation", "out/checkpoints/tp10"))

Quelle est la bonne explication ?

• a) En mode append, une fenêtre n'est écrite qu'une fois close par le watermark, et ces fenêtres ne l'étaient pas encore
• b) Toutes les lignes de ces minutes ont été ignorées par le watermark parce qu'elles étaient en retard
• c) Le format Parquet n'accepte pas les fenêtres incomplètes, il faudrait écrire en CSV
• d) Le checkpoint a gardé ces fenêtres en mémoire et ne les écrira jamais

Question 8

Un collègue veut compter en continu le nombre exact d'utilisateurs distincts sur le flux de logs avec count(distinct user). Que lui réponds-tu ?

• a) C'est possible, à condition de passer en outputMode("append")
• b) C'est possible, à condition d'ajouter un watermark de 2 minutes
• c) C'est possible, mais uniquement avec trigger(availableNow=True)
• d) Le distinct exact n'est pas supporté sur un flux, mais approx_count_distinct l'est

Question 9

Dans le TP 10, tu veux afficher les logs triés par ts. Tu ajoutes .orderBy("ts") directement sur le flux logs parsé, sans agrégation, puis tu écris en append. Quelle est la bonne approche ?

• a) Spark refuse ce tri sur un flux sans agrégation : on trie après coup, en lisant le Parquet en batch comme à l'étape 5
• b) Le tri fonctionne, mais seulement à l'intérieur de chaque micro-batch
• c) Le tri fonctionne si tu ajoutes option("maxFilesPerTrigger", 1) pour respecter l'ordre d'arrivée
• d) Le tri fonctionne si tu passes en outputMode("update")

Question 10

Bonus du TP 9 : après 60 secondes de stream en direct, le calcul en batch sur les mêmes fichiers donne plus de courses que le stream. Le simulateur produit 1 fichier toutes les 2 secondes, la requête traite maxFilesPerTrigger = 2 fichiers par micro-batch de 5 secondes. Quel diagnostic retiens-tu ?

• a) Le stream perd des courses à cause de la jointure flux-statique, qui écarte des zones
• b) Le stream a écarté les 2 % de courses en retard, car il applique un watermark par défaut
• c) Le stream a pris du retard sur le simulateur ; laissé plus longtemps, il rattrape les fichiers et retrouve les mêmes comptes que le batch
• d) Le mode complete ne conserve que les derniers micro-batches, donc l'écart restera définitif

Question 11

Tu lances python scripts/stream_simulator.py courses --all --reset avec les valeurs par défaut (30 fichiers de 500 courses). Puis tu exécutes le corrigé du TP 9 en mode availableNow, avec maxFilesPerTrigger = 2. Que montre q.recentProgress à la fin ?

• a) Un seul micro-batch d'environ 15 000 courses, car availableNow lit d'un coup tout ce qui est disponible
• b) 15 micro-batches d'environ 1 000 courses chacun, car maxFilesPerTrigger reste appliqué
• c) 30 micro-batches de 500 courses, car Spark traite toujours un fichier par micro-batch
• d) Aucun micro-batch, car availableNow n'attend que les fichiers arrivés après le démarrage

Question 12

Pour aller plus vite, un participant retire la ligne .schema(schema) de son readStream du TP 9 :

courses = (spark.readStream
           .option("maxFilesPerTrigger", 2)
           .json("stream/courses"))

Que se passe-t-il ?

• a) Spark infère le schéma sur le premier fichier, exactement comme spark.read.json
• b) Le flux est créé, mais toutes les colonnes sont de type chaîne
• c) Le flux ne contient qu'une colonne value, comme avec readStream.text
• d) Spark refuse de créer le flux : une source fichier en streaming exige un schéma explicite

Question 13

Avec courses --reset, le simulateur vide stream/courses, mais supprime aussi out/checkpoints/tp9 et out/checkpoints/tp11. Pourquoi ?

• a) Parce que les checkpoints occupent trop de place sur un portable de formation
• b) Parce que les checkpoints contiennent une copie des fichiers de courses, devenue inutile
• c) Parce qu'ils mémorisent les offsets et l'état d'un flux qui n'existe plus : la requête reprendrait l'ancien état et ignorerait les nouveaux fichiers
• d) Parce que les TP 9 et 11 écrivent leurs résultats dans stream/courses

Question 14

Dans le TP 10, tu oublies le withWatermark et tu démarres cette requête :

sans_wm = (logs
           .groupBy(F.window("ts", "1 minute"), "endpoint")
           .agg(F.count("*").alias("requetes")))
q = (sans_wm.writeStream
     .outputMode("append")
     .format("parquet")
     .option("path", "out/tp10/par_minute")
     .option("checkpointLocation", "out/checkpoints/tp10")
     .start())

Que se passe-t-il ?

• a) Spark refuse la requête : une agrégation en mode append exige un watermark
• b) La requête démarre, mais n'écrit jamais rien dans le Parquet
• c) La requête réécrit toutes les fenêtres dans le Parquet à chaque micro-batch
• d) Spark ajoute de lui-même un watermark de 2 minutes, et tout fonctionne

Question 15

À l'étape 6 du TP 10, q.lastProgress affiche un watermark 2024-01-15T13:27:55.000Z. Les logs du simulateur sont pourtant datés de la matinée. Comment lis-tu cette valeur ?

• a) Le watermark est en avance de 5 heures : les lignes entre 8h27 et 13h27 seront toutes ignorées
• b) Le simulateur a produit des logs datés de 13h27, après la matinée
• c) Le watermark a dépassé toutes les fenêtres, donc le Parquet contient toute la matinée
• d) C'est 8h27 à New York : lastProgress affiche le watermark en UTC

Question 16

Après un logs --reset, tu lis le Parquet du TP 10. Une fenêtre qui commence à 7h52 apparaît, alors que l'horloge du simulateur démarre à 8h00. Quelle explication retiens-tu ?

• a) to_timestamp a mal lu certaines lignes, et leur heure a été décalée
• b) Ce sont des lignes arrivées avec 8 minutes de retard au tout début du flux, acceptées car le watermark n'avait pas encore avancé
• c) Spark a converti ces lignes dans un autre fuseau horaire au moment de la fenêtre
• d) Ce sont des lignes d'une ancienne exécution restées dans stream/logs

Question 17

Ton équipe veut deux choses : une alerte sur /payment la plus réactive possible, et un bilan exact des erreurs de la matinée, sans ligne perdue. Quelle approche choisis-tu ?

• a) Un seul stream avec un watermark de 10 minutes, qui sert à la fois pour l'alerte et pour le bilan
• b) Un stream sans watermark, pour ne jamais ignorer de ligne en retard
• c) Un watermark court, 2 minutes, pour l'alerte, et un batch de rattrapage sur les logs stockés pour le bilan exact
• d) Un watermark de 0 minute, pour que chaque fenêtre soit écrite dès sa première ligne

Question 18

Dans le bonus du TP 10, tu remplaces la fenêtre de 1 minute par F.window("ts", "5 minutes", "1 minute"). Quelle affirmation est juste ?

• a) Les fenêtres de 5 minutes ne se chevauchent pas, et l'alerte réagit 5 fois plus vite
• b) Spark ne garde qu'une minute sur 5, pour réduire le volume traité
• c) La fenêtre glissante s'appuie sur l'heure d'arrivée des lignes plutôt que sur ts
• d) Chaque événement compte dans 5 fenêtres : le signal est lissé, l'alerte monte plus progressivement mais résiste mieux aux pics isolés

Question 19

Les alertes du TP 10 (taux d'erreur > 10 %) montrent /payment aux minutes 08:22, 08:23, 08:24 et 08:25. Le simulateur indique pourtant une panne « entre 8h22 et 8h26 ». Un collègue s'étonne de ne pas voir 08:26. Quelle est la bonne explication ?

• a) La colonne minute vient de window.start : la fenêtre 08:25 couvre 8h25 à 8h26, donc la panne tient dans les 4 fenêtres de 08:22 à 08:25
• b) La fenêtre 08:26 a été entièrement ignorée par le watermark
• c) Le taux d'erreur de 08:26 est passé juste sous 10 % à cause des lignes en retard
• d) La fenêtre 08:26 n'est pas encore close, comme les fenêtres de 8h27 à 8h29

Question 20

Ton voisin a oublié la ligne spark.conf.set("spark.sql.shuffle.partitions", "4") dans le TP 9. Sa requête donne les bons totaux par Borough, mais chaque micro-batch est nettement plus lent que chez toi. Quel diagnostic retiens-tu ?

• a) Les cœurs de local[*] ne suffisent pas, il faut demander 200 cœurs
• b) Sans cette ligne, l'agrégation répartit chaque petit micro-batch sur 200 partitions par défaut, soit beaucoup de tâches presque vides
• c) Le checkpoint n'accepte que 4 partitions, et Spark le réécrit à chaque micro-batch
• d) Spark lit 200 fichiers par micro-batch au lieu de respecter maxFilesPerTrigger

Corriger Recommencer

Flashcards

Cliquez sur une carte pour la retourner. Mélangez-les pour une révision active.

Question Que renvoie courses.isStreaming juste après un spark.readStream...json("stream/courses") ?

◀ Précédente Retourner Suivante ▶ Mélanger 1 / 20