Module 4 — Lecture de données et formats colonnes

Le module précédent a insisté sur l'importance de pousser les filtres au niveau du stockage. Encore faut-il que le stockage sache les recevoir. C'est là que Parquet entre en scène, et pourquoi remplacer un CSV par un Parquet est souvent l'optimisation qui rapporte le plus, avant toute autre.

CSV contre Parquet, sur un même jeu

Un fichier CSV est un texte ligne par ligne, sans schéma, sans compression native, orienté ligne. Pour lire une seule colonne parmi trente, Spark doit tout de même lire la ligne entière ; pour filtrer sur une date, il doit tout lire jusqu'à la trouver, puisqu'il n'y a pas d'index.

Un fichier Parquet est un binaire orienté colonne, compressé par colonne, avec des statistiques par bloc (min, max, count) et un schéma embarqué. Pour lire trois colonnes sur trente, Spark lit littéralement trois colonnes : la taille lue chute d'un facteur dix. Pour filtrer sur une plage de dates, il consulte le min/max de chaque bloc et saute les blocs hors plage : c'est ce que fait PushedFilters du plan d'exécution du module 3.

Sur notre jeu de vols, la différence tient de la caricature. Un CSV de 28 Go se lit en dix minutes ; le même contenu en Parquet compressé occupe 6 Go et se lit en une minute. Toutes choses égales par ailleurs.

La première question à poser

Devant un CSV massif à traiter plusieurs fois, la première action utile est presque toujours de le convertir en Parquet partitionné une fois pour toutes. Le coût de conversion est amorti dès la deuxième lecture.

Le piège du schéma inféré

Spark peut deviner le schéma d'un CSV avec inferSchema=True. La commande paraît anodine :

vols = (
    spark.read
    .option("header", True)
    .option("inferSchema", True)
    .csv("data/vols.csv")
)

Le coût caché : pour deviner les types, Spark fait un premier passage complet sur les données, uniquement pour lire. Sur 28 Go de CSV, ces dix minutes sont dépensées avant même la première ligne utile. Puis un second passage suit pour le vrai calcul.

La bonne pratique consiste à donner explicitement le schéma :

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType

schema = StructType([
    StructField("compagnie", StringType(), nullable=False),
    StructField("origine", StringType(), nullable=False),
    StructField("destination", StringType(), nullable=False),
    StructField("date_depart", TimestampType(), nullable=False),
    StructField("retard_arrivee", IntegerType(), nullable=True),
])

vols = spark.read.schema(schema).option("header", True).csv("data/vols.csv")

Un seul passage, types corrects, valeurs manquantes explicites plutôt que promues en StringType par défaut. En Parquet, le schéma est déjà dans le fichier : spark.read.parquet(...) suffit.

Partitionnement par colonne

Un Parquet partitionné stocke ses lignes dans une arborescence par valeur de colonne :

data/vols/
  annee=2023/mois=01/part-00000.parquet
  annee=2023/mois=02/part-00000.parquet
  ...
  annee=2024/mois=12/part-00000.parquet

Un filtre sur annee = 2024 ne visite que les répertoires annee=2024/, sans même ouvrir les fichiers des autres années : c'est l'élagage de partitions (partition pruning). Le gain est mécanique : traiter un mois sur cent millions de lignes revient à lire 1/12ᵉ de ce que Spark lirait sans partition.

L'écriture partitionnée se fait par partitionBy :

(
    vols
    .write
    .mode("overwrite")
    .partitionBy("annee", "mois")
    .parquet("data/vols_parquet")
)

Attention à ne pas partitionner sur une colonne à trop forte cardinalité : partitionner par date_depart créerait un fichier par jour, chaque fichier deviendrait minuscule, et Spark perdrait plus de temps à ouvrir les fichiers qu'à lire leur contenu. Une règle raisonnable : viser des partitions de 128 Mo à 1 Go chacune.

Lecture du fil rouge

Voici la lecture standard du jeu de vols pour tous les modules suivants :

vols = (
    spark.read
    .parquet("data/vols_parquet")
    .filter(F.col("annee").between(2019, 2024))
    .filter(F.col("retard_arrivee").isNotNull())
)

print(vols.count())                # doit afficher plusieurs dizaines de millions
vols.select("compagnie", "origine", "retard_arrivee").show(5)

Les deux filtres tirent parti de l'élagage (années) et de Parquet (colonne non nulle). Un count() sur Parquet est presque gratuit : les statistiques par bloc donnent le nombre de lignes sans les lire.

Autres formats à connaître

• ORC : très proche de Parquet, très utilisé dans l'écosystème Hive. Choisir selon ce que consomme votre stockage aval, la performance est comparable.
• Delta / Iceberg / Hudi : couches transactionnelles au-dessus de Parquet, avec versionnement et écritures atomiques ; c'est le défaut sur Databricks (Delta) et de plus en plus dans l'entrepôt de données.
• JSON : pratique pour ingérer des journaux, coûteux à parcourir. Convertir en Parquet dès que possible.

Le nombre de fichiers compte

Un dossier Parquet avec dix mille petits fichiers de 4 Mo se lit beaucoup plus lentement qu'un dossier avec deux cents fichiers de 200 Mo, même si le total est identique : Spark passe son temps à ouvrir et fermer. Le module 8 revient sur cette question dite du « petit fichier ».

En résumé

• Parquet est colonnaire, compressé, avec un schéma et des statistiques par bloc ; CSV n'a rien de tout cela.
• inferSchema=True fait un passage complet sur les données avant tout calcul : lourd sur gros volume, à éviter.
• Le partitionnement par colonne autorise l'élagage ; ne partitionner que sur des colonnes à faible cardinalité.
• Trop de petits fichiers coûte cher : viser des partitions de 128 Mo à 1 Go, à revoir au module 8.

Module suivant : les briques de base de Spark ML, transformateurs et estimateurs, pour préparer les données à l'entraînement.