Ce notebook présente les opérations courantes sur les DataFrames Spark : lecture d’un fichier CSV, sélection et conversion de colonnes, agrégations, fonctions de fenêtre et jointures.
Exécutez les cellules dans l’ordre. La cellule de préparation télécharge directement le jeu de données des transactions depuis GitHub.
Objectifs
À la fin du notebook, vous saurez :
décrire le rôle de Spark SQL et la structure d’un DataFrame ;
choisir et vérifier les types des colonnes ;
sélectionner, transformer et agréger des données ;
distinguer une agrégation par groupe d’un calcul par fenêtre ;
interpréter une jointure en tenant compte des clés et des doublons.
Parcours
Spark SQL et les DataFrames
RDD et types de données
Charger et préparer les transactions
Sélectionner des colonnes
Ajouter des colonnes avec withColumn
Agréger avec groupBy
Calculer avec des fonctions de fenêtre
Joindre des DataFrames
# Spark 3.5.7 prend en charge Java 8, 11 ou 17.!pip install -q pyspark==3.5.7!java -version
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv[notice] A new release of pip is available: 24.0 -> 26.2.1[notice] To update, run: pip install --upgrade pip
openjdk version "17.0.20.1" 2026-08-18
OpenJDK Runtime Environment (build 17.0.20.1+1-1-deb12u1-Debian)
OpenJDK 64-Bit Server VM (build 17.0.20.1+1-1-deb12u1-Debian, mixed mode, sharing)
# Créer ou récupérer la session Spark du notebook.spark = ( SparkSession.builder .master("local[*]") .appName("CM3-SparkSQL-DataFrames") .getOrCreate())spark.sparkContext.setLogLevel("ERROR")
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
26/10/15 23:05:37 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
1. Spark SQL et les DataFrames
Spark SQL est le module de Spark consacré au traitement des données structurées. Il permet de décrire un même calcul avec l’API DataFrame ou avec SQL. Spark analyse et optimise le plan avant de l’exécuter ; les effets de ces optimisations dépendent des données, des opérations et de l’environnement.
Un DataFrame est une collection distribuée de lignes organisées en colonnes nommées et typées. Il peut s’exécuter sur une seule machine, comme ici en mode local, ou sur un cluster.
# Construire un petit DataFrame avec un schéma explicite.data_livres = [ (1, "Le Petit Prince", "Antoine de Saint-Exupéry", date(1943, 4, 6), Decimal("12.50"), "Fiction"), (2, "1984", "George Orwell", date(1949, 6, 8), Decimal("15.00"), "Dystopie"), (3, "L'Étranger", "Albert Camus", date(1942, 5, 19), Decimal("10.00"), "Philosophie"), (4, "Harry Potter à l'école des sorciers", "J.K. Rowling", date(1997, 6, 26), Decimal("20.00"), "Fantastique"), (5, "Les Misérables", "Victor Hugo", date(1862, 4, 3), Decimal("18.00"), "Drame"), (6, "Le Seigneur des Anneaux", "J.R.R. Tolkien", date(1954, 7, 29), Decimal("25.00"), "Fantastique"), (7, "To Kill a Mockingbird", "Harper Lee", date(1960, 7, 11), Decimal("14.00"), "Fiction"), (8, "Crime et Châtiment", "Fiodor Dostoïevski", date(1866, 1, 1), Decimal("16.50"), "Psychologique"), (9, "Pride and Prejudice", "Jane Austen", date(1813, 1, 28), Decimal("13.00"), "Romance"), (10, "Moby Dick", "Herman Melville", date(1851, 10, 18), Decimal("17.00"), "Aventure"),]schema_livres = StructType([ StructField("id", IntegerType(), nullable=False), StructField("livre", StringType(), nullable=False), StructField("auteur", StringType(), nullable=False), StructField("date_publication", DateType(), nullable=False), StructField("prix_eur", DecimalType(10, 2), nullable=False), StructField("genre", StringType(), nullable=False),])df_livres = spark.createDataFrame(data_livres, schema_livres)df_livres.printSchema()df_livres.toPandas()
Un DataFrame reprend le principe d’une collection distribuée, comme un RDD, et y associe un schéma tabulaire ainsi qu’un plan de requête optimisable par Catalyst.
On peut retenir cette formule pédagogique :
DataFrame ≈ RDD + schéma en colonnes + optimisations de Catalyst
Le schéma décrit notamment les noms et les types des colonnes. Les opérations comme select, filter, groupBy ou join décrivent le calcul sous une forme que Spark SQL peut analyser. Un DataFrame représente ainsi une table distribuée dont le moteur connaît la structure et les opérations demandées.
Les opérations construisent un plan logique, qui décrit le résultat demandé. Spark SQL s’appuie sur Catalyst, son optimiseur de requêtes, pour préparer le traitement :
Analyser le plan : identifier les colonnes et vérifier la compatibilité des types.
Optimiser le plan logique : simplifier les expressions, supprimer les colonnes inutiles ou déplacer les filtres lorsque cela préserve le résultat.
Préparer le plan physique : choisir les opérateurs qui réaliseront le calcul, notamment les stratégies de jointure.
Le plan logique décrit les opérations demandées ; le plan physique précise leur réalisation. La commande df.explain(mode="extended") permet d’observer ces différentes étapes.
Utiliser un RDD lorsque :
Le traitement manipule des données non structurées ou semi-structurées sous forme d’objets, par exemple des collections hétérogènes, des arbres ou des graphes, dont la manipulation directe convient mieux à l’algorithme.
Le traitement nécessite un algorithme distribué personnalisé, par exemple une méthode de factorisation matricielle ou un nouvel algorithme de machine learning dont on définit les calculs et les échanges de données.
Privilégier un DataFrame lorsque :
Les données peuvent être décrites par un schéma, avec des colonnes nommées et typées, par exemple des transactions, des événements ou des documents JSON.
Le traitement repose sur des opérations tabulaires : sélections, filtres, agrégations, jointures ou fonctions de fenêtre.
Les données doivent être interrogées en SQL, après leur enregistrement comme vue temporaire.
Le calcul peut être exprimé avec les fonctions intégrées de Spark, afin de bénéficier des optimisations de Catalyst.
Types courants
Le type d’une colonne détermine les valeurs qu’elle peut représenter et les opérations disponibles.
Type PySpark
Rôle
Exemple
ByteType
Entier signé sur 8 bits, de −128 à 127
ByteType()
ShortType
Entier signé sur 16 bits, de −32 768 à 32 767
ShortType()
IntegerType
Entier signé sur 32 bits, de −2 147 483 648 à 2 147 483 647
IntegerType()
LongType
Entier signé sur 64 bits, de −9 223 372 036 854 775 808 à 9 223 372 036 854 775 807
LongType()
FloatType
Nombre binaire à virgule flottante simple précision
FloatType()
DoubleType
Nombre binaire à virgule flottante double précision
DoubleType()
DecimalType(precision, scale)
Décimal à précision et nombre de décimales définis
DecimalType(10, 2)
StringType
Texte
"PySpark"
BooleanType
Valeur vraie ou fausse
True
DateType
Date sans heure
date(2023, 10, 1)
TimestampType
Horodatage tenant compte du fuseau horaire de la session
"2023-10-01 12:34:56"
TimestampNTZType
Horodatage sans fuseau horaire
"2023-10-01 12:34:56"
ArrayType, MapType
Collection de valeurs ou association clé-valeur
["a", "b"], {"a": 1}
3. Charger et préparer les transactions
La cellule suivante télécharge transactions_10000.csv dans l’espace de travail temporaire de Google Colab. Le fichier doit contenir les colonnes transaction_id, customer_id, product_id, product_categorie, boutique, amount et date.
Avec header=True, la première ligne fournit les noms des colonnes. Avec inferSchema=False, Spark conserve les valeurs des colonnes CSV sous forme de chaînes ; nous les convertirons explicitement dans la suite.
# Lire le CSV : sans inférence de schéma, les colonnes sont des chaînes.df = ( spark.read .option("header", True) .option("inferSchema", False) .csv("transactions_10000.csv"))df.printSchema()
select projette un sous-ensemble de colonnes ou d’expressions dans un nouveau DataFrame. Cette projection permet de ne conserver que les données utiles au calcul.
Affichons quelques lignes avec limit(5). toPandas() convient ici parce que le résultat est volontairement petit ; cette méthode rapatrie les lignes sélectionnées dans la mémoire du driver.
df.select("boutique", "date").limit(5).toPandas()
boutique
date
0
Auchan - Toulouse Capitole
2024-06-18
1
Auchan - Toulouse Capitole
2024-06-18
2
Auchan - Toulouse Capitole
2024-06-18
3
Monoprix - Paris Opéra
2024-09-12
4
Monoprix - Paris Opéra
2024-09-12
select accepte aussi des expressions de colonnes. On peut notamment :
renommer une colonne avec alias ;
convertir son type avec cast ;
calculer une expression ;
appliquer une fonction intégrée comme lower, min ou sum.
# Renommage de la colonne boutiquedf.select(F.col("boutique").alias("BOUTIQUE"), "date").limit(10).toPandas()
BOUTIQUE
date
0
Auchan - Toulouse Capitole
2024-06-18
1
Auchan - Toulouse Capitole
2024-06-18
2
Auchan - Toulouse Capitole
2024-06-18
3
Monoprix - Paris Opéra
2024-09-12
4
Monoprix - Paris Opéra
2024-09-12
5
Monoprix - Paris Opéra
2024-09-12
6
Monoprix - Paris Opéra
2024-09-12
7
Monoprix - Paris Opéra
2024-09-12
8
Franprix - Paris Bastille
2024-03-15
9
Auchan - Toulouse Capitole
2024-03-19
# Convertir les identifiants, le montant et la date ; normaliser les textes.df_prepared = df.select( F.col("transaction_id").cast("bigint").alias("transaction_id"), F.col("customer_id").cast("bigint").alias("customer_id"), F.col("product_id").cast("bigint").alias("product_id"), F.lower(F.col("product_categorie")).alias("product_categorie"), F.lower(F.col("boutique")).alias("boutique"), F.col("amount").cast(DecimalType(18, 2)).alias("amount"), F.to_date(F.col("date"), "yyyy-MM-dd").alias("date"),)df_prepared.printSchema()df_prepared.limit(10).toPandas()
groupBy définit les clés de regroupement. L’appel à agg calcule ensuite une ou plusieurs mesures et produit une ligne par groupe. Un regroupement peut nécessiter un échange de données (shuffle) afin de réunir les lignes qui partagent la même clé.
Dans cet exemple, chaque ligne du résultat correspond à une boutique.
Une fonction de fenêtre calcule une valeur à partir des lignes d’un même groupe tout en conservant une ligne par observation. partitionBy définit les groupes de la fenêtre ; orderBy définit l’ordre utilisé par les fonctions de rang, lag et lead.
# Données d'exemple pour les fonctions de rang et de décalage.data_employes = [ ("Ventes", "Alice", 7000), ("Ventes", "Bob", 6000), ("Ventes", "Charlie", 6000), ("Ventes", "Dave", 5000), ("RH", "Eve", 6500), ("RH", "Frank", 6500), ("RH", "Grace", 5500),]colonnes_employes = ["Département", "Employé", "Salaire"]df_employes = spark.createDataFrame(data_employes, colonnes_employes)# Définition des fenêtresw_rang = Window.partitionBy("Département").orderBy(F.desc("Salaire"))w_ordre = Window.partitionBy("Département").orderBy(F.desc("Salaire"), F.asc("Employé"))w_departement = Window.partitionBy("Département")resultat_fenetres = ( df_employes .withColumn("numero_ligne", F.row_number().over(w_ordre)) .withColumn("rang", F.rank().over(w_rang)) .withColumn("rang_dense", F.dense_rank().over(w_rang)) .withColumn("salaire_precedent", F.lag("Salaire").over(w_ordre)) .withColumn("salaire_suivant", F.lead("Salaire").over(w_ordre)) .withColumn("total_departement", F.sum("Salaire").over(w_departement)) .orderBy("Département", F.desc("Salaire"), "Employé"))resultat_fenetres.toPandas()
Département
Employé
Salaire
numero_ligne
rang
rang_dense
salaire_precedent
salaire_suivant
total_departement
0
RH
Eve
6500
1
1
1
NaN
6500.0
18500
1
RH
Frank
6500
2
1
1
6500.0
5500.0
18500
2
RH
Grace
5500
3
3
2
6500.0
NaN
18500
3
Ventes
Alice
7000
1
1
1
NaN
6000.0
24000
4
Ventes
Bob
6000
2
2
2
7000.0
6000.0
24000
5
Ventes
Charlie
6000
3
2
2
6000.0
5000.0
24000
6
Ventes
Dave
5000
4
4
3
6000.0
NaN
24000
row_number attribue un numéro distinct à chaque ligne selon l’ordre de la fenêtre. Les ex æquo sont ici départagés par le nom de l’employé.
rank donne le même rang aux salaires égaux et laisse un saut dans la suite des rangs.
dense_rank donne le même rang aux ex æquo sans laisser de saut.
lag et lead lisent le salaire de la ligne précédente ou suivante dans l’ordre défini. Sans critère de départage, l’ordre entre salaires égaux n’est pas déterminé.
La somme sur w_departement, qui ne définit pas d’ordre, calcule le total du département et le répète sur chacune de ses lignes.
8. Joindre des DataFrames
Une jointure associe les lignes de deux DataFrames à partir d’une condition ou d’une clé commune. Avec on="Étudiant_ID", Spark compare les valeurs de cette colonne dans les deux DataFrames.
Type
Lignes conservées
inner
Les lignes dont la clé apparaît dans les deux DataFrames
left
Toutes les lignes de gauche et les correspondances à droite
right
Toutes les lignes de droite et les correspondances à gauche
full
Toutes les lignes des deux côtés, avec des valeurs nulles en l’absence de correspondance
cross
Toutes les combinaisons de lignes des deux DataFrames
Une clé répétée peut produire plusieurs correspondances et multiplier les lignes. Un étudiant inscrit à deux cours apparaît deux fois dans une jointure gauche. Vérifiez la cardinalité des clés avant de calculer des sommes après la jointure.
