Partie 4 – Structured API et Spark SQL

• 15h15 – 17h00
• Apports : 15 min
• TP 4 : 45 min
• Partage + récap : 30 min

Et si le TP 2 tenait en 5 lignes de SQL ? Comment Spark s'y prend-il pour exécuter du SQL sur des téraoctets ?

Spark SQL : de SQLContext et HiveContext à SparkSession

Hive a été le premier à proposer du SQL sur Hadoop, en traduisant les requêtes en jobs MapReduce. Spark SQL l’a remplacé pour le calcul : même SQL, moteur Spark. Il peut toujours lire les tables du metastore Hive.

Hier (Spark 1.x)	Aujourd’hui (Spark 4)
SQLContext : requêtes SQL sur des DataFrames	SparkSession : un seul point d’entrée
HiveContext : SQL + tables du metastore Hive	.enableHiveSupport() pour le metastore Hive
Deux objets de plus à gérer avec le SparkContext	Le même optimiseur (Catalyst) pour SQL et DataFrames
Encore présents dans du code ancien : savoir les lire	spark.sql("SELECT …") ou API DataFrame : même performance

Démonstration à faire : la même requête en SQL et en API DataFrame, puis explain() sur les deux → même plan physique.

DataFrames : structure, schéma et partitionnement

Un DataFrame est un ensemble de lignes avec des colonnes nommées et typées, distribué comme un RDD mais optimisé par Catalyst. Techniquement, c’est un Dataset[Row].

      RDD (lignes brutes non structurées)               DataFrame (table structurée avec schéma)
 ┌─────────────────────────────────────────┐       ┌────────┬──────────┬──────────┐
 │ "id;nom;prenom"                         │       │ Id     │ nom      │ prenom   │
 │ "1;toto;yoyo"                           │  ──▶  ├────────┼──────────┼──────────┤
 │ "2;titi;jiji"                           │       │ 1      │ toto     │ yoyo     │
 │ "3;tata;gogo"                           │       │ 2      │ titi     │ jiji     │
 │ "4;tutu;nono"                           │       │ 3      │ tata     │ gogo     │
 └─────────────────────────────────────────┘       │ 4      │ tutu     │ nono     │
                                                   └────────┴──────────┴──────────┘

Critère (Feature)	RDD	DataFrame	Dataset
Immuable (Immutable)	Oui	Oui	Oui
Tolérant aux pannes (Fault tolerant)	Oui	Oui	Oui
Typage fort à la compilation (Type-safe)	Oui	Non (vérifié à l’exécution)	Oui (Scala / Java)
Schéma de colonnes (Schema)	Non	Oui	Oui
Optimiseur d’exécution (Catalyst)	Non	Oui	Oui
Niveau d’abstraction (Level)	Bas (Low)	Haut (High)	Haut (High)

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, IntegerType, DoubleType, TimestampNTZType

schema = StructType([
    StructField("PULocationID", IntegerType()),
    StructField("tpep_pickup_datetime", TimestampNTZType()),
    StructField("total_amount", DoubleType()),
])
df = (spark.read.schema(schema).parquet("data/trips.parquet")
        .withColumn("jour", F.to_date("tpep_pickup_datetime")))

df.rdd.getNumPartitions()                      # combien de partitions en mémoire ?
(df.write.mode("overwrite")                    # sans overwrite, la 2e exécution échoue : le dossier existe
   .partitionBy("jour").parquet("out/trips_par_jour"))   # un dossier jour=… par valeur (TP 4, étape 5)

Notion	Ce qu’il faut savoir
Structure	Des lignes et des colonnes nommées et typées. printSchema() les affiche.
Schéma explicite	Le déclarer évite à Spark de relire les données pour le deviner (inferSchema=True crée des jobs avant votre count()). Attention aux types : une colonne Parquet INT64 exige LongType, pas IntegerType. L’horodatage NYC TLC est en heure locale sans fuseau : TimestampNTZType.
Partitionnement en mémoire	repartition(n) redistribue (shuffle) ; coalesce(n) réduit le nombre de partitions sans shuffle.
Partitionnement sur disque	partitionBy("jour") range les fichiers par valeur : les requêtes filtrées sur jour ne lisent que les dossiers utiles. Combien de fichiers ? Au plus partitions en mémoire × valeurs distinctes.

🔍 Vulgarisation : repartition(n) vs coalesce(n) vs partitionBy("jour") — ne plus jamais les confondre

Ces trois méthodes contiennent le mot « partition », mais font des choses très différentes :

1. df.repartition(8) (en mémoire, AVEC shuffle complet) :
   • Mélange toutes les lignes sur le réseau pour créer 8 partitions parfaitement équilibrées en RAM.
   • Quand l’utiliser ? Pour augmenter le parallélisme ou rééquilibrer des partitions tordues avant un gros calcul.
2. df.coalesce(2) (en mémoire, SANS shuffle) :
   • Regroupe simplement des partitions existantes sur la même machine pour passer par exemple de 8 à 2 partitions, sans échanger les données sur le réseau.
   • Quand l’utiliser ? Juste avant d’écrire le résultat sur disque pour éviter de générer 200 minuscules fichiers.
3. df.write.partitionBy("jour").parquet("out/") (sur disque, arborescence de dossiers) :
   • Crée un sous-dossier physique par date : out/jour=2024-01-01/part-000.parquet, out/jour=2024-01-02/...
   • Gain majeur (Partition Pruning) : quand quelqu’un écrit WHERE jour = '2024-01-15', Spark ouvre uniquement le dossier jour=2024-01-15 et ignore les 30 autres jours (31× moins de lecture disque !).

Pourquoi Parquet plutôt que CSV ?

CSV	Parquet
Texte, ligne par ligne	Binaire, par colonnes
Pas de schéma : inférence coûteuse ou déclaration	Schéma embarqué
Lit toutes les colonnes même pour en utiliser 3	Ne lit que les colonnes demandées
Pas de compression ni de statistiques	Compressé, statistiques min / max par bloc, lecture vectorisée

Le TP 11 (jour 3) mesurera un gain de 2 à 3 fois sur la même requête.

🧮 Sous le capot : Comment Parquet lit 10× moins de données sur disque grâce au stockage en colonnes et aux statistiques Min/Max ?

Regardons comment 3 courses avec 19 colonnes sont écrites physiquement sur le disque dur :

En CSV (stockage par LIGNE) :
  Disque : [id1, date1, pass1, dist1, ..., total1] [id2, date2, pass2, dist2, ..., total2] ...
  👉 Pour calculer SUM(total_amount), la tête de lecture est obligée de traverser les 19 colonnes de CHAQUE ligne !

En Parquet (stockage par COLONNE + statistiques par bloc) :
  Disque : ── Colonne PULocationID : [132, 132, 48, ...] (compressé)
           ── Colonne trip_distance : [18.2, 1.4, 3.1, ...] (min=0.5, max=24.0)
           ── Colonne total_amount  : [70.0, 12.5, 19.0, ...]
           ── (les 16 autres colonnes sont rangées plus loin et ne sont MÊME PAS LUES !)

Deux super-pouvoirs de Parquet :

1. Column Pruning (élagage des colonnes) : si votre requête utilise 2 colonnes sur 19 (PULocationID et total_amount), Spark ne lit sur le disque que 2/19e des colonnes (environ 10 % du fichier).
2. Predicate Pushdown avec statistiques Min/Max : chaque bloc Parquet note dans son en-tête le min et le max de chaque colonne. Si vous filtrez WHERE trip_distance > 50 et que l’en-tête du bloc indique max(trip_distance) = 24.0, Spark saute le bloc entier sans même le décompresser !

💡 Vulgarisation : Le Broadcast Join expliqué en 30 secondes (pourquoi le TP 4 évite un gros shuffle)

Vous devez joindre 120 000 courses (trips, plusieurs mégaoctets ou gigaoctets) avec le référentiel des 265 zones de taxi (zones, seulement 12 Ko).

• Jointure classique (Sort-Merge Join, comme dans les RDD du TP 2) : Spark trie et envoie les 120 000 courses sur le réseau (shuffle) pour que les courses de la zone 132 rencontrent le libellé de la zone 132. Très coûteux !
• Broadcast Hash Join (choisi automatiquement par Catalyst en TP 4) : Catalyst remarque que zones est minuscule (< 10 Mo). Au lieu de déplacer les 120 000 courses, le Driver photocopie la petite table des 265 zones et l’envoie une fois à chaque Executor (broadcast). Chaque Executor fait alors la jointure sur place en RAM, avec 0 shuffle de la grande table !

Catalyst, en une phrase (le reste au jour 3)

Catalyst transforme votre requête – SQL, Python ou Scala – en un plan logique, l’optimise (filtres poussés vers la lecture, colonnes élaguées, choix du type de jointure) puis choisit un plan physique. C’est pourquoi le join du TP 4 avec le petit référentiel des zones devient un Broadcast join (explain()), et pourquoi Python va aussi vite que Scala avec les DataFrames : Python ne fait que construire le plan, la JVM l’exécute.

À vous de jouer

→ TP 4 – Le fil rouge en Spark SQL (45 min)

Partage et remédiation

• RDD (TP 2) ou DataFrame (TP 4) : lequel est le plus rapide ? Le plus lisible ? Le DataFrame, sur les deux plans. Catalyst optimise (broadcast join, lecture des seules colonnes utiles dans Parquet) ; le SQL se lit en une fois.
• Quand utiliser encore les RDD ? Données non structurées (texte brut, binaire), contrôle fin du partitionnement, code existant. Dans les autres cas : DataFrames d’abord.
• Tour de table : la notion la plus claire et la plus floue de la journée. Les notions floues alimentent les questions-réponses de demain matin.

Ce qu’il faut retenir de la journée

1. Diviser pour régner : partitions, calcul là où sont les données, en parallèle et en mémoire.
2. Paresseux jusqu’à l’action : transformations = un plan ; 1 action = 1 job, 1 shuffle = 1 stage.
3. Un seul point d’entrée : SparkSession. SQLContext et HiveContext = héritage à savoir lire.
4. Qui fait quoi : Driver planifie, Cluster Manager alloue, Executors exécutent, HDFS stocke par blocs.
5. DataFrames d’abord : schéma explicite, SQL ou API, Catalyst optimise. Les RDD pour les cas spéciaux.

Cliffhanger pour demain : « Demain, on prédit le montant d’une course avant qu’elle ne commence. » → Jour 2 – Machine Learning avec Spark ML