Module 3 — Transformations, actions et évaluation paresseuse

Le module précédent a montré que Spark construit un plan avant d'exécuter. La suite logique est de comprendre ce qui construit ce plan (les transformations) et ce qui le déclenche (les actions). Deux développeurs sur trois qui trouvent Spark « lent » ont en fait un plan qui se recalcule trois fois : ce module leur explique pourquoi.

Transformations et actions

Une transformation décrit une nouvelle vue des données à partir d'une autre : select, filter, withColumn, groupBy, join, orderBy. Ces appels ne déclenchent aucun calcul ; ils ajoutent une étape au plan logique. Le DataFrame retourné est un objet léger qui contient le plan, pas les données.

Une action demande un résultat concret : show, count, collect, take, write. Là et seulement là, Spark compile le plan, expédie les tâches aux exécuteurs, et attend le résultat.

retards_cdg = vols.filter(F.col("origine") == "CDG")   # transformation
retards_gros = retards_cdg.filter(F.col("retard_arrivee") > 15)  # transformation
compte = retards_gros.count()  # action, déclenche l'exécution

Cette évaluation paresseuse (lazy evaluation) est le fondement des optimisations Catalyst. Si Spark exécutait le premier filter immédiatement, il matérialiserait un DataFrame intermédiaire ; en attendant l'action, il fusionne les deux filtres et pousse le tout à la lecture Parquet.

Étroit ou large : ce qui coûte cher

Toutes les transformations ne se valent pas. Une transformation étroite (narrow) traite chaque partition indépendamment : select, filter, withColumn, map. Une partition d'entrée produit exactement une partition de sortie, sans échange entre exécuteurs. Le coût réseau est nul.

Une transformation large (wide) exige que les lignes voyagent entre exécuteurs : groupBy, join, distinct, orderBy, repartition. Ces opérations provoquent un brassage (shuffle) : chaque exécuteur écrit ses données sur disque, un tri partitionné les redistribue selon la clé, et les exécuteurs relisent leur nouvelle partition. Sur un jeu de plusieurs gigaoctets, un brassage coûte de dix à cent fois plus qu'une transformation étroite.

D'où une règle d'or : éviter le brassage inutile. Filtrer avant de grouper, joindre sur une table diffusée en mémoire (broadcast join) quand elle est petite, ne pas trier si un min ou un max suffit.

Le piège du recalcul

L'évaluation paresseuse a un revers. Chaque action recalcule le DataFrame depuis la source, sauf indication contraire. Considérez :

retards = vols.filter(F.col("retard_arrivee") > 15)
print(retards.count())          # première lecture du Parquet
retards.show(10)                # deuxième lecture du Parquet
retards.write.parquet("out")    # troisième lecture du Parquet

Trois lectures du fichier source, trois passes de calcul. Sur cent millions de lignes, c'est deux passes de trop.

cache et persist

Pour matérialiser un DataFrame en mémoire des exécuteurs et le réutiliser sans le recalculer, on appelle cache() (raccourci pour persist(MEMORY_AND_DISK)) ou persist(niveau) :

retards = vols.filter(F.col("retard_arrivee") > 15).cache()
retards.count()                 # matérialise dans le cache
retards.show(10)                # relit le cache, pas le Parquet
retards.write.parquet("out")    # relit le cache
retards.unpersist()             # libère la mémoire

Quelques précautions. Le cache() n'a d'effet qu'après la première action qui matérialise le DataFrame : appeler cache() seul ne fait rien. Les niveaux MEMORY_ONLY, MEMORY_AND_DISK, DISK_ONLY arbitrent entre vitesse et robustesse : MEMORY_ONLY perd la partition si la mémoire manque, MEMORY_AND_DISK déborde sur disque. Enfin, garder au cache un DataFrame utilisé une seule fois est du gaspillage — on n'y met que ce qu'on relit plusieurs fois.

Lire un plan d'exécution

.explain() sans argument donne le plan physique ; .explain(True) donne aussi les plans analysés, optimisés et physiques ; .explain(mode="formatted") produit une vue arborescente lisible :

== Physical Plan ==
* HashAggregate(keys=[compagnie], functions=[avg(retard_arrivee)])
+- Exchange hashpartitioning(compagnie, 16)      <-- brassage
   +- * HashAggregate(keys=[compagnie], functions=[partial_avg(retard_arrivee)])
      +- * Filter (isnotnull(origine) AND (origine = CDG))
         +- * ColumnarToRow
            +- FileScan parquet [compagnie,origine,retard_arrivee]
                 PushedFilters: [IsNotNull(origine), EqualTo(origine, CDG)]

Les indices utiles sont partout. PushedFilters confirme que le filtre est descendu au fichier — c'est pour cela que Parquet est si efficace. Exchange hashpartitioning marque le brassage. HashAggregate en deux temps (partial_avg puis avg) montre l'agrégation locale avant la globale. Un plan qui contient plusieurs Exchange là où vous n'en attendiez qu'un est un signe fort d'un orderBy ou d'un join mal placé.

L'ordre des transformations compte

vols.select("compagnie", "retard_arrivee").filter(F.col("compagnie") == "AF") et vols.filter(F.col("compagnie") == "AF").select("compagnie", "retard_arrivee") produisent le même résultat, mais Catalyst optimise en poussant le filtre : dans les deux cas, seule la colonne compagnie est lue puis filtrée au niveau du fichier. Vérifier, ne pas croire — lire explain reste le seul juge.

En résumé

• Les transformations décrivent un plan sans rien exécuter ; seules les actions (count, show, write) déclenchent le calcul.
• Les transformations étroites restent locales à la partition, les larges provoquent un brassage coûteux ; en éviter est la première optimisation à faire.
• Chaque action relit la source par défaut ; cache() matérialise le DataFrame après la première action et amortit les suivantes.
• .explain() révèle PushedFilters, Exchange, et l'ordre réel des opérations ; c'est le premier outil de diagnostic à ouvrir.

Module suivant : la lecture proprement dite, où CSV et Parquet ne jouent pas dans la même catégorie.