TP 2 – Top 10 des zones avec les RDD

• 30 min · 11h10 – 11h40
• Notebook : jour1/TP2_rdd_top10_zones.ipynb
• Données : data/trips.csv, data/taxi_zone_lookup.csv

📁 Dans le dépôt

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

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

Objectif : trouver les 10 zones de départ (PULocationID) qui rapportent le plus (total_amount), avec l’API RDD. Colonnes de trips.csv (index à partir de 0) : PULocationID = colonne 7, total_amount = colonne 16.

Livrable : le notebook enregistré dans le dépôt, sur votre branche (git checkout -b prenom, puis git add / git commit).

Réussi si : le top 10 affiche les noms des zones, JFK Airport en tête (≈ 261 823 $).

Étapes

1. Charger data/trips.csv en RDD de lignes et retirer l’en-tête

Indice : sc.textFile(...), first() renvoie la première ligne, puis filter(lambda l: l != entete). Chaque élément du RDD est une chaîne : une ligne complète, non découpée.

2. Construire un PairRDD (PULocationID, total_amount)

Indice : map(lambda l: l.split(",")) puis map(lambda c: (int(c[7]), float(c[16]))). Vérifiez avec take(3), pas avec collect() (qui rapatrierait les 120 000 paires sur le Driver).

3. Chiffre d’affaires par zone avec reduceByKey

reduceByKey(lambda a, b: a + b). Le résultat compte une ligne par zone présente dans le fichier.

4. Garder le top 10 avec takeOrdered

Indice : takeOrdered(10, key=lambda kv: -kv[1]). takeOrdered renvoie les plus petites valeurs de la clé de tri, d’où le signe moins. C’est une action : top10 est une liste Python.

5. Afficher le nom des zones : join avec data/taxi_zone_lookup.csv

Le fichier met tous les champs entre guillemets : préférez csv.reader à un split(",").

import csv
champs = next(csv.reader([ligne]))   # ['132', 'Queens', 'JFK Airport', 'Airports']

• Retirez l’en-tête du fichier des zones avant de convertir.
• Les clés doivent avoir le même type des deux côtés : int(c[0]). Avec '132' (texte) d’un côté et 132 (entier) de l’autre, le join renvoie un résultat vide, sans aucune erreur.
• Construisez un PairRDD (LocationID, Zone), puis sc.parallelize(top10).join(zones). On réduit d’abord (10 lignes), on enrichit ensuite.
• Le join ne garde pas l’ordre : retriez avec sortBy(lambda kv: -kv[1][0]).

6. Dans la Spark UI : combien de stages ? Où est le shuffle ?

Écrivez votre réponse dans le notebook. Attendu : 2 stages pour le top 10 (lecture + map, puis agrégation après le shuffle de reduceByKey). Le join et le sortBy ajoutent des stages supplémentaires.

Bonus – groupByKey contre reduceByKey

Refaites l’étape 3 avec groupByKey().mapValues(sum) et lancez un count() sur chaque version. Dans la Spark UI, onglet Stages, comparez la colonne Shuffle Write : ≈ 480 Ko pour groupByKey, ≈ 7 Ko pour reduceByKey, soit environ 70 fois plus de données envoyées sur le réseau. Pourquoi ? reduceByKey additionne d’abord sur chaque partition et n’envoie qu’un total par zone.

Ne comparez pas les temps : sur 120 000 lignes et un portable, la différence est noyée dans le bruit. Sur un vrai cluster, c’est le volume du shuffle qui coûte.

Correction (live coding)

lignes = sc.textFile("data/trips.csv")
entete = lignes.first()                              # la première ligne est l'en-tête du CSV
courses = lignes.filter(lambda l: l != entete)

ca = (courses.map(lambda l: l.split(","))
             .map(lambda c: (int(c[7]), float(c[16])))
             .reduceByKey(lambda a, b: a + b))
top10 = ca.takeOrdered(10, key=lambda kv: -kv[1])

import csv
zl = sc.textFile("data/taxi_zone_lookup.csv")
zl_entete = zl.first()
zones = (zl.filter(lambda l: l != zl_entete)
           .map(lambda l: next(csv.reader([l])))
           .map(lambda c: (int(c[0]), f"{c[2]} ({c[1]})")))
resultat = sc.parallelize(top10).join(zones).sortBy(lambda kv: -kv[1][0])
for zid, (montant, nom) in resultat.collect():
    print(nom, round(montant, 2))

Pièges rencontrés en séance

Symptôme	Cause	Solution
join vide, sans erreur	Clés '132' d’un côté, 132 de l’autre	int(c[0]) des deux côtés ; retirer l’en-tête des zones
ValueError: invalid literal for int() with base 10: '"132"' au collect()	split(",") laisse les guillemets ; les map sont paresseux, l’erreur sort à l’action	csv.reader (ou replace('"', ''))
Le top 10 donne les zones qui rapportent le moins	Signe moins oublié dans takeOrdered	key=lambda kv: -kv[1]
Relire trips.csv à chaque action	Normal : Spark ne conserve pas les RDD intermédiaires	cache() si on enchaîne plusieurs actions

Partage

• Un participant présente sa solution (tiré au sort en classe virtuelle) ; correction en direct.
• Pourquoi reduceByKey est-il plus rapide que groupByKey ? Pré-agrégation par partition avant le shuffle.
• Combien de stages, et pourquoi ? Un nouveau stage à chaque transformation wide.
• À retenir : 1 shuffle = 1 nouveau stage. C’est là que se joue la performance.