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].

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.

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.

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