Quiz 10 – ML en streaming, Catalyst, Tungsten, AQE et architectures Lambda / Kappa

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

Contenus évalués : application d'un `PipelineModel` sauvegardé sur un flux avec `transform` et `foreachBatch` ; préparation des features commune au batch et au streaming, colonne `probability` et taux prédit ; triggers `availableNow` et `processingTime`, watermark sur un horodatage converti ; lecture d'un plan Catalyst avec `explain("formatted")`, `PushedFilters` et `ReadSchema` ; Tungsten, whole-stage codegen et lecture vectorisée Parquet ; AQE, plan final, coalescence des partitions de shuffle et jointures adaptatives ; choix entre Lambda, Kappa et architecture hybride selon le besoin.

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

Au TP 11, tu recharges le modèle de pourboire du TP 7 avec PipelineModel.load("out/modele_pourboire"). Quelle instruction applique ce modèle sur le flux de courses ?

• a) modele.fit(features(flux))
• b) modele.transform(features(flux))
• c) modele.predict(flux), après avoir converti le flux en DataFrame batch
• d) PipelineModel.load(...) à chaque micro-batch, puis modele.fit(batch_df)

Question 2

Tu lances requete(csv).explain("formatted") avec requete = lambda df: df.filter("trip_distance > 10").groupBy("PULocationID").agg(F.avg("total_amount")). Quelle ligne du plan montre que le filtre trip_distance > 10 est envoyé directement à la source de données ?

• a) ReadSchema
• b) Batched
• c) PushedFilters
• d) AQEShuffleRead

Question 3

Dans un plan physique, certains opérateurs sont préfixés par *(1), *(2), etc. Que signifie ce préfixe ?

• a) Ces opérateurs sont compilés ensemble en bytecode par Tungsten (whole-stage codegen).
• b) Le nombre de partitions traitées par l'opérateur.
• c) Le nombre d'exécutions de la requête, ici une moyenne sur 3.
• d) Le numéro du job dans l'onglet Jobs de la Spark UI.

Question 4

Après r.collect(), le plan affiché par r.explain() contient AQEShuffleRead coalesced. Qu'a fait AQE ?

• a) Il a poussé le filtre dans la lecture du fichier Parquet.
• b) Il a remplacé le shuffle par une lecture vectorisée.
• c) Il a fixé spark.sql.shuffle.partitions à 200, la valeur par défaut.
• d) Il a fusionné les petites partitions de shuffle pendant l'exécution.

Question 5

Lors de l'atelier, un client demande un tableau de bord rafraîchi toutes les 5 secondes et des alertes en cas d'anomalie. Quelle architecture correspond à la correction proposée ?

• a) Lambda : un job batch chaque nuit sur les données archivées, qui recalcule le tableau de bord.
• b) Kappa : Kafka puis Structured Streaming.
• c) Un job batch sur Parquet relancé toutes les heures.
• d) Lambda : une seule base de code batch, sans couche temps réel.

Question 6

Tu compares explain("formatted") en CSV puis en Parquet. Dans les deux cas, PushedFilters contient le filtre et ReadSchema ne garde que 3 colonnes. Qu'en conclus-tu ?

• a) Catalyst applique les mêmes optimisations logiques quelle que soit la source ; la différence de performance est physique, elle vient du format.
• b) Le pushdown de filtre ne fonctionne qu'en Parquet ; en CSV, PushedFilters est affiché mais ignoré.
• c) Les deux lectures auront exactement le même temps d'exécution, puisque les plans sont identiques.
• d) AQE a réécrit le plan CSV pour qu'il ressemble au plan Parquet.

Question 7

Au TP 11, la même requête prend environ 0,42 s en CSV et 0,14 s en Parquet, soit trois fois moins. Quelle explication est juste ?

• a) Catalyst pousse le filtre en Parquet mais pas en CSV.
• b) Parquet active AQE, alors que CSV le désactive.
• c) Parquet est un format colonne : il saute les colonnes inutiles sur le disque, utilise les statistiques min/max pour sauter des blocs et lit en vectorisé (Batched: true).
• d) Le fichier Parquet contient moins de courses que le CSV.

Question 8

Tu lances la requête de suivi : watermark de 10 minutes, fenêtres de 30 minutes, outputMode("append") et foreachBatch(traiter). Le premier micro-batch n'écrit aucune fenêtre. Quel est le diagnostic ?

def traiter(batch_df, batch_id):
    batch_df.write.mode("append").parquet("out/tp11/suivi")
    n = batch_df.count()
    if n:
        print(f"micro-batch {batch_id} : {n} fenêtres closes écrites")

• a) foreachBatch n'est pas compatible avec le mode append ; il faut passer en complete.
• b) Le modèle n'a pas été rechargé, donc la colonne p_pourboire est vide.
• c) L'ancien dossier de checkpoint bloque l'écriture ; il faut le supprimer.
• d) C'est normal : en append, une fenêtre n'est écrite qu'une fois close par le watermark, et aucune ne l'est encore.

Question 9

Tu exécutes r = requete(pq) puis r.collect(). Ensuite, tu lances requete(pq).explain() et tu lis isFinalPlan=false. Pourquoi ?

• a) AQE est désactivé par défaut dans Spark 4.
• b) requete(pq) crée un nouvel objet DataFrame, jamais exécuté ; le plan final n'existe qu'après l'exécution, sur le même objet.
• c) explain() sans argument n'affiche jamais le plan final ; il faut explain("formatted").
• d) La lecture Parquet empêche AQE de finaliser le plan.

Question 10

Le même client ajoute un besoin : une facturation mensuelle exacte, à la course près. Le tableau de bord et les alertes passent déjà par le flux. Quelle approche retiens-tu ?

• a) Une architecture hybride : le flux pour le tableau de bord et les alertes, un job batch sur les données archivées (Parquet / Delta) pour la facturation.
• b) Tout en streaming : il suffit d'un watermark très court pour que la facturation soit exacte.
• c) Tout en batch : on abandonne le tableau de bord à 5 secondes.
• d) Tout en streaming en mode append : chaque ligne n'est écrite qu'une fois, la facturation est donc exhaustive.

Question 11

Au TP 11, toute la préparation des features (filtre carte bancaire, heure, jointure avec zones, manhattan, nuit, label) tient dans une seule fonction features(df). Elle sert à l'entraînement quand le modèle manque, et sur le flux. Pourquoi ce choix ?

• a) Parce que transform n'accepte qu'une fonction Python en argument, pas un DataFrame.
• b) Parce qu'un DataFrame en streaming refuse withColumn en dehors d'une fonction.
• c) Pour que les colonnes soient calculées exactement de la même façon à l'entraînement et à la prédiction : aucun décalage possible entre batch et streaming.
• d) Pour que Catalyst garde en cache la jointure avec zones et ne la recalcule plus à chaque micro-batch.

Question 12

Dans foreachBatch(traiter), la fonction traiter(batch_df, batch_id) écrit batch_df en Parquet puis appelle batch_df.count(). Qu'est-ce que batch_df ?

• a) Un DataFrame batch ordinaire, qui contient le résultat de la requête pour ce micro-batch ; on peut y appliquer n'importe quel code batch, comme write.parquet ou count.
• b) Une seule ligne du flux : traiter est appelée une fois par course reçue.
• c) Le flux complet depuis le début, relu en entier à chaque micro-batch.
• d) Un DataFrame en streaming : count() y est interdit, il faudrait passer par writeStream.

Question 13

Tu as lancé python scripts/stream_simulator.py courses --all --reset, puis le corrigé sans définir la variable MODE. Que fait la requête ?

flux = spark.readStream.schema(schema).option("maxFilesPerTrigger", 3).json("stream/courses")
...
q = (q.trigger(processingTime="5 seconds") if MODE == "live" else q.trigger(availableNow=True)).start()
q.awaitTermination(120 if MODE == "live" else None)

• a) Elle tourne sans fin et relit stream/courses toutes les 5 secondes.
• b) Elle traite un seul fichier, puis s'arrête.
• c) Elle attend 120 secondes puis s'arrête, quel que soit le nombre de fichiers traités.
• d) Elle traite tous les fichiers déjà présents, par micro-batches de 3 fichiers au plus, puis s'arrête d'elle-même.

Question 14

Un participant écrit predictions.withWatermark("tpep_pickup_datetime", "10 minutes") et obtient une erreur au lancement. Quel est le diagnostic ?

• a) Le watermark doit être au moins égal à la durée de la fenêtre, soit 30 minutes.
• b) La colonne est de type TIMESTAMP_NTZ (sans fuseau), que le watermark refuse : il faut le poser sur ts, obtenue par F.col("tpep_pickup_datetime").cast("timestamp").
• c) withWatermark doit être appelé après le groupBy, pas avant.
• d) Le watermark n'est pas compatible avec foreachBatch ; il faut un sink Parquet direct.

Question 15

Au TP 7, un participant a sauvegardé le LinearSVC dans out/modele_pourboire. Au TP 11, le chargement réussit, mais la ligne vector_to_array("probability")[1] échoue. Que fais-tu ?

• a) Ajouter .withColumn("probability", F.col("prediction")) avant vector_to_array.
• b) Remplacer [1] par [0] : le LinearSVC range la probabilité de la classe 1 en première position.
• c) Réentraîner le Random Forest du TP 7 et le sauvegarder à cet endroit : le LinearSVC ne produit pas de colonne probability.
• d) Recharger le modèle avec LinearSVC.load au lieu de PipelineModel.load.

Question 16

Le corrigé compte les blocs whole-stage codegen du plan final ainsi :

plan_final = r._jdf.queryExecution().executedPlan().toString()
len(set(re.findall(r"\*\((\d+)\)", plan_final)))

Pourquoi passer par set ?

• a) Pour trier les numéros dans l'ordre d'exécution des stages.
• b) Parce que findall renvoie aussi les numéros des partitions de shuffle, qu'il faut éliminer.
• c) Pour écarter les opérateurs qui ne sont pas compilés par Tungsten.
• d) Plusieurs opérateurs d'un même bloc portent le même préfixe *(n) : on compte les numéros distincts, donc les blocs de code générés, et non les opérateurs.

Question 17

Avant l'exécution, Catalyst choisit une stratégie de jointure à partir d'estimations de taille. Qu'apporte AQE sur une jointure ?

• a) Il peut changer de stratégie pendant l'exécution, à partir des tailles réellement mesurées après le shuffle, par exemple diffuser (broadcast) un côté qui s'est révélé petit.
• b) Il pousse la condition de jointure dans la lecture du fichier, comme un PushedFilters.
• c) Il remplace la jointure par une lecture vectorisée des deux tables.
• d) Il fige la stratégie avant l'exécution, à partir du ReadSchema des deux côtés.

Question 18

Dans la requête de suivi, taux_predit est calculé avec F.avg("p_pourboire") et non avec F.avg("prediction"). Pourquoi ?

• a) La colonne prediction n'existe pas sur un DataFrame en streaming.
• b) La moyenne d'une colonne 0/1 est interdite dans une agrégation avec watermark.
• c) La moyenne des probabilités estime la proportion attendue de courses avec pourboire, comparable à taux_reel ; la moyenne des 0/1 ne compte que les courses au-dessus du seuil, et une fenêtre où toutes les probabilités valent 0,6 afficherait 100 %.
• d) Les probabilités sont toujours égales au label, donc l'écart prédit / réel devient nul.

Question 19

Lors du partage, tu constates que taux_predit et taux_reel font des sauts d'une fenêtre à l'autre pour un arrondissement qui compte peu de courses. Quelle approche retiens-tu pour réduire ce bruit ?

• a) Raccourcir le watermark à 2 minutes pour publier les fenêtres plus vite.
• b) Allonger les fenêtres, agréger plusieurs quartiers, ou afficher un intervalle de confiance.
• c) Passer la requête en outputMode("complete").
• d) Remplacer la moyenne de p_pourboire par la moyenne de prediction.

Question 20

Pour le tableau de bord à 5 secondes et les alertes, un collègue propose Lambda : une couche batch et une couche temps réel en parallèle. Quel argument justifie de préférer Kappa pour ce besoin ?

• a) Lambda ne permet pas d'alimenter un tableau de bord.
• b) Kappa garantit une facturation exacte sans aucun job batch.
• c) Kappa n'a pas besoin de checkpoint, contrairement à Lambda.
• d) Kappa n'a qu'un seul chemin de traitement et une seule base de code ; Lambda impose de maintenir deux traitements, batch et temps réel, et de garder leurs résultats cohérents.

Corriger Recommencer

Flashcards

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

Question Appliquer un PipelineModel sauvegardé sur un flux (TP 11)

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