SparkSession et DataFrames Spark

Quand pandas ne suffit plus

pandas charge toutes les données dans la mémoire d’une seule machine. Au-delà de quelques dizaines de gigaoctets, ou quand un traitement doit lire des milliers de fichiers d’un lac de données, il faut répartir le travail. Apache Spark est le moteur de calcul distribué le plus répandu pour cela : il découpe les données en partitions et les traite en parallèle sur plusieurs machines, en mémoire autant que possible. On l’utilise en Python avec PySpark. Ce cours s’appuie sur l’API DataFrame de Spark 3.5 et 4.x.

Spark n’est pas toujours la bonne réponse : pour quelques millions de lignes, pandas ou DuckDB sur une machine sont plus simples et souvent plus rapides. Spark devient pertinent quand le volume ou le parallélisme l’exige.

L’architecture en trois rôles

• Le driver exécute votre script, construit le plan de calcul et distribue le travail.
• Les executors sont des processus lancés sur les machines du cluster ; chacun traite des tâches, une tâche correspondant à une partition de données.
• Le gestionnaire de cluster (Kubernetes, YARN, le mode autonome de Spark, ou un service managé) attribue les ressources.

En local, tout tourne dans un seul processus (master("local[*]") utilise tous les cœurs) : pratique pour apprendre, avec exactement la même API.

Créer une session et lire des données

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .appName("ventes")
         .master("local[*]")          # en cluster, ce paramètre est fourni au lancement
         .getOrCreate())

schema = "id INT, jour DATE, region STRING, produit STRING, montant DOUBLE"

ventes = (spark.read
          .option("header", True)
          .option("mode", "DROPMALFORMED")   # ignorer les lignes illisibles
          .schema(schema)
          .csv("s3a://lake/brut/ventes/"))   # un dossier entier : tous ses fichiers

ventes.printSchema()
ventes.show(5)

Le schéma explicite évite deux problèmes de inferSchema=True : une lecture supplémentaire des données pour deviner les types, et des types devinés faux (un code postal lu comme entier perd son zéro initial). Le mode de lecture décide du sort des lignes mal formées : PERMISSIVE (par défaut) les garde en mettant à null ce qui ne se lit pas, DROPMALFORMED les écarte, FAILFAST arrête la lecture à la première erreur.

Pour les données déjà préparées, préférez Parquet : format en colonnes, compressé, qui embarque son schéma (spark.read.parquet(...)).

Transformer un DataFrame

L’API ressemble à pandas, mais avec un style déclaratif inspiré de SQL :

resultat = (ventes
    .filter(F.col("montant") > 0)
    .withColumn("ttc", F.round(F.col("montant") * 1.2, 2))
    .withColumn("gamme", F.when(F.col("montant") >= 200, "haut").otherwise("standard"))
    .select("jour", "region", "produit", "ttc", "gamme"))

Trois règles :

1. Un DataFrame est immuable : withColumn ne modifie pas ventes, il renvoie un nouveau DataFrame. Il faut donc réaffecter ou enchaîner.
2. Les expressions sur colonnes (F.col, F.when, F.to_date, F.upper…) sont exécutées dans le moteur de Spark, optimisées et distribuées. Une fonction Python appliquée ligne à ligne (UDF) est bien plus lente : réservez-la aux cas sans équivalent natif.
3. Rien n’est encore calculé à ce stade : Spark attend une action (show, count, write…), comme on le verra à la leçon suivante.

Ramener des données vers le driver

show(n) affiche quelques lignes ; count() compte ; collect() rapatrie toutes les lignes dans la mémoire du driver. Sur un vrai volume, collect() fait tomber le driver : préférez limit(n), show(), ou écrivez le résultat (write.parquet). Pour passer à pandas un résultat déjà agrégé et petit, toPandas() convient.

À retenir

• Driver = chef d’orchestre, executors = ouvriers, une tâche par partition.
• Schéma explicite et mode de lecture choisi ; Parquet pour les données préparées.
• DataFrame immuable, fonctions F plutôt que Python ligne à ligne, et jamais de collect() sur un gros volume.