TP 4 – Le fil rouge en DataFrames et Spark SQL

• 45 min · 15h30 – 16h15
• Notebook : jour1/TP4_spark_sql.ipynb
• Données : data/trips.parquet, data/taxi_zone_lookup.csv

📁 Dans le dépôt

• Notebook : jour1/TP4_spark_sql.ipynb
• Données : data/trips.csv, data/taxi_zone_lookup.csv
• Corrigé : jour1/corriges/tp4_sql.py (publié en fin de J1)

À exécuter depuis la racine du dépôt. Pas encore cloné ? Voir Prérequis et installation.

Objectif : refaire le TP 2 avec l’API structurée, aller plus loin (heure, arrondissement), et comparer avec les RDD.

Réussi si : le top 10 SQL est identique à celui du TP 2, et 31 dossiers jour=… sont écrits.

Étapes

1. Lire trips.parquet avec un schéma explicite

Gardez seulement tpep_pickup_datetime, PULocationID, DOLocationID, trip_distance, total_amount.

Indices : StructType([StructField(nom, type), ...]), spark.read.schema(schema).parquet(...). Les identifiants sont des entiers 32 bits (IntegerType), les montants des DoubleType, l’horodatage est en heure locale sans fuseau (TimestampNTZType). Un type faux lève PARQUET_COLUMN_DATA_TYPE_MISMATCH : Parquet ne convertit pas.

2. Créer les vues trips et zones

Le référentiel des zones est un petit CSV avec en-tête ; ici inferSchema=True est acceptable (263 lignes) :

zones = spark.read.csv("data/taxi_zone_lookup.csv", header=True, inferSchema=True)

createOrReplaceTempView("trips") puis createOrReplaceTempView("zones") : des vues temporaires, visibles dans spark.sql, qui vivent avec la session. La jointure se fait sur t.PULocationID = z.LocationID.

3a. En SQL : le top 10 des zones de départ, avec le nom de la zone et l’arrondissement

Indice : JOIN zones z ON t.PULocationID = z.LocationID, GROUP BY, ORDER BY ... DESC LIMIT 10.

3b. En SQL : courses et chiffre d’affaires par heure. À quelle heure New York rapporte-t-il le plus ?

Indice : HOUR(tpep_pickup_datetime) AS heure, COUNT(*), SUM(total_amount), ORDER BY heure. Attendu : 18 h (≈ 237 600 $), puis 19 h et 17 h.

4. Avec l’API DataFrame : classement des arrondissements par chiffre d’affaires

Indice : join, groupBy("Borough"), agg(F.sum(...)), orderBy(F.desc(...)). Même résultat que du SQL, même plan : c’est la même chose pour Catalyst.

5. Écrire les courses en Parquet, partitionnées par jour

Indice : withColumn("jour", F.to_date(...)), write.mode("overwrite").partitionBy("jour").parquet("out/trips_par_jour"). Combien de dossiers obtenez-vous ? (31 : un par jour de janvier.)

Bonus 1 – explain()

Affichez le plan physique du top 10. Que fait Catalyst avec la jointure ? (BroadcastHashJoin : le petit référentiel des zones est envoyé à tous les Executors, pas de shuffle de la grande table.) Comparez la lisibilité avec le TP 2, et le temps : t = time.time(); top10.collect(); print(time.time() - t).

Bonus 2 – ANSI SQL dans Spark 4

SELECT total_amount / trip_distance FROM trips WHERE trip_distance = 0

Que se passe-t-il ? En Spark 4, le mode ANSI est activé par défaut : DIVIDE_BY_ZERO, là où Spark 3 renvoyait NULL en silence. Corrigez avec try_divide(total_amount, trip_distance).

Attention : sans le WHERE, un simple show() ne lit que les 20 premières lignes, qui n’ont pas de distance nulle : vous ne verriez aucune erreur. Spark est paresseux jusqu’au bout.

Correction (extraits)

from pyspark.sql import functions as F
from pyspark.sql.types import *

schema = StructType([
    StructField("tpep_pickup_datetime", TimestampNTZType()),
    StructField("PULocationID", IntegerType()),
    StructField("DOLocationID", IntegerType()),
    StructField("trip_distance", DoubleType()),
    StructField("total_amount", DoubleType()),
])
trips = spark.read.schema(schema).parquet("data/trips.parquet")
trips.createOrReplaceTempView("trips")
zones.createOrReplaceTempView("zones")

top10 = spark.sql("""
    SELECT z.Zone, z.Borough, ROUND(SUM(t.total_amount), 2) AS ca
    FROM trips t JOIN zones z ON t.PULocationID = z.LocationID
    GROUP BY z.Zone, z.Borough
    ORDER BY ca DESC LIMIT 10
""")

(trips.withColumn("jour", F.to_date("tpep_pickup_datetime"))
      .write.mode("overwrite").partitionBy("jour").parquet("out/trips_par_jour"))

Partage

• RDD ou DataFrame : lequel est le plus rapide ? Le plus lisible ? Pourquoi ? Le DataFrame : Catalyst choisit un broadcast join et ne lit que les colonnes utiles du Parquet ; cinq lignes de SQL remplacent vingt lignes de RDD.
• Que change --master dans spark-submit ? Où s’exécutent les Executors ; le code métier ne change pas.
• Tour de table : la notion la plus claire et la plus floue de la journée.