Fiche mémo – Jour 1

En une minute

Spark fait du calcul distribué en mémoire. Il a remplacé Hadoop MapReduce grâce à des API unifiées en Python, Scala, Java et R. Tout part d’une SparkSession. Rien ne se calcule avant une action : les transformations ne font que construire un plan, le DAG. Le même code métier tourne en local[*] sur votre portable ou sur un cluster Standalone avec HDFS : seuls --master et le chemin des données changent. Les DataFrames et Spark SQL font le même travail que les RDD en moins de lignes, et plus vite, car Catalyst optimise le plan.

Notions clés

Notion	Définition en une ligne	Piège fréquent
SparkSession	Point d’entrée unique, qui contient le SparkContext (spark.sparkContext).	Croire qu’il faut créer un SparkContext à part. SQLContext et HiveContext sont de l’héritage.
local[*]	Exécution locale avec autant de threads que de cœurs.	Laisser .master() dans un script lancé par spark-submit : le master doit venir de la ligne de commande.
Spark UI	Interface de l’application, sur le port 4040.	Port déjà pris, Spark passe à 4041 : le bon lien est spark.sparkContext.uiWebUrl.
RDD	Collection distribuée, immuable, découpée en partitions.	Croire qu’on le modifie : chaque transformation crée un nouveau RDD.
Lignée	Historique des transformations d’un RDD.	Elle sert à recalculer les partitions d’un Executor tombé. Ce n’est pas une copie des données.
Transformation / action	map, filter, reduceByKey, join sont paresseuses. count, collect, take, saveAsTextFile déclenchent le calcul.	inferSchema=True relit le fichier : des jobs apparaissent avant votre count().
Narrow / wide	Narrow : sans échange réseau. Wide : avec shuffle (reduceByKey, join).	Penser que map ou filter créent un stage.
Job / stage / tâche	1 action = 1 job. 1 stage par shuffle, plus 1. 1 tâche par partition.	Compter les opérations au lieu de compter les shuffles.
reduceByKey / groupByKey	reduceByKey pré-agrège sur chaque partition avant le shuffle.	Utiliser groupByKey().mapValues(sum), qui envoie toutes les valeurs sur le réseau.
join sur PairRDD	Associe deux PairRDD par clé.	Clés '132' d’un côté et 132 de l’autre : résultat vide, sans erreur. Le join ne garde pas l’ordre.
Driver / Executor	Le Driver exécute votre programme et planifie le DAG. Les Executors, hébergés par les Workers, exécutent les tâches.	Oublier que chaque Worker doit voir les données. D’où HDFS ou un stockage objet.
Cluster Manager	Alloue les ressources. Standalone est intégré à Spark ; YARN et Kubernetes sont les alternatives. Mesos a été retiré en 4.0.	Croire que changer de gestionnaire oblige à réécrire le code.
HDFS	Le NameNode garde les métadonnées, les DataNodes stockent les blocs répliqués.	Confondre les fichiers : core-site.xml dit où est le système de fichiers (fs.defaultFS), hdfs-site.xml dit comment stocker (réplication, blocs).
DataFrame / Catalyst	Données en colonnes typées. Catalyst optimise le plan (broadcast join, projection poussée vers Parquet).	Penser que SQL et API DataFrame diffèrent en performance : même moteur.
Schéma explicite	StructType fixé à la lecture, sans inférence.	Mettre IntegerType sur une colonne Parquet INT64 lève PARQUET_COLUMN_DATA_TYPE_MISMATCH.
Parquet	Format colonne, compressé, schéma embarqué.	Croire que CSV et Parquet se valent : CSV impose l’inférence du schéma.
ANSI SQL	Activé par défaut dans Spark 4 : une division par zéro lève une erreur.	show() ne lit que 20 lignes, donc l’erreur peut ne pas apparaître. Filtrer, ou utiliser try_divide.

Autres nouveautés de Spark 3 et 4 citées dans le programme : Spark Connect, le type VARIANT et AQE. AQE est détaillé au jour 3.

Code à connaître

Des extraits, pas des programmes complets : chacun suppose les imports et les variables du corrigé indiqué en commentaire (jour1/corriges/ dans le dépôt). Imports communs :

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

# tp1_premier_contact.py – une session locale, tous les cœurs, et le lien de la Spark UI
spark = SparkSession.builder.appName("TP").master("local[*]").getOrCreate()
sc = spark.sparkContext
print(sc.uiWebUrl)

# tp1_premier_contact.py – charger et inspecter (inferSchema relit le fichier)
df = spark.read.csv("data/trips.csv", header=True, inferSchema=True)
df.printSchema(); df.show(5); df.count()
df.rdd.getNumPartitions()            # = nombre de tâches du premier stage

# tp2_rdd.py – RDD : CA par zone, puis top 10 (colonnes 7 et 16)
lignes = sc.textFile("data/trips.csv")
entete = lignes.first()
courses = lignes.filter(lambda l: l != entete)
paires = courses.map(lambda l: l.split(",")).map(lambda c: (int(c[7]), float(c[16])))
top10 = paires.reduceByKey(lambda a, b: a + b).takeOrdered(10, key=lambda kv: -kv[1])

# tp2_rdd.py – join : même type de clé des deux côtés, puis on retrie ; valeur = « Zone (Borough) »
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])))          # csv.reader : les zones contiennent des virgules
           .map(lambda c: (int(c[0]), f"{c[2]} ({c[1]})")))
sc.parallelize(top10).join(zones).sortBy(lambda kv: -kv[1][0])

# jour1/TP3_cluster/tp2_cluster.py – même script, autre cluster : seul --master change
spark-submit --master spark://127.0.0.1:7077 jour1/TP3_cluster/tp2_cluster.py data

# tp4_sql.py – schéma explicite (les 5 colonnes utiles) + SQL
schema = StructType([
    StructField("tpep_pickup_datetime", TimestampNTZType()),   # heure locale sans fuseau
    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")
spark.sql("SELECT PULocationID, SUM(total_amount) AS ca FROM trips GROUP BY PULocationID")

# tp4_sql.py – écriture partitionnée, et division sûre en ANSI
trips.withColumn("jour", F.to_date("tpep_pickup_datetime")) \
     .write.mode("overwrite").partitionBy("jour").parquet("out/trips_par_jour")
spark.sql("SELECT try_divide(total_amount, trip_distance) FROM trips")

Chiffres des TP

• Données : 120 000 courses, 19 colonnes, 1 mois. 265 zones dans taxi_zone_lookup.csv. PULocationID en colonne 7, total_amount en colonne 16 (index à partir de 0).
• TP 1 : le job du count() a 2 stages. Spark UI sur le port 4040, ou 4041 s’il est pris.
• TP 2 : JFK Airport en tête, ≈ 261 823 $.
• TP 2, bonus : shuffle de groupByKey ≈ 480 Ko, de reduceByKey ≈ 7 Ko : environ 70 fois moins.
• TP 3, cluster : master sur spark://…:7077, son interface sur le port 8080. Interface HDFS sur 9870, RPC du NameNode sur 8020. Le worker a 2 cœurs.
• TP 3, HDFS : trips.csv fait 12 Mo, moins qu’un bloc de 128 Mo : 1 seul bloc. Réplication 1 en TP, 3 en production.
• TP 4 : le chiffre d’affaires culmine à 18 h (≈ 237 600 $), puis 19 h et 17 h. L’écriture partitionnée produit 31 dossiers jour=….
• Prérequis : Java 17+, Python 3.10+, PySpark 4.x, 8 Go de RAM minimum (16 Go pour le TP 3 avec Docker). Sous Windows, WSL2. check_env.py vérifie aussi une vraie écriture Parquet.

Questions que pose un formateur

1. Pourquoi des jobs apparaissent-ils avant votre count() ? inferSchema=True relit le fichier pour deviner les types (et header=True lit la première ligne).
2. Combien de stages pour map().filter().reduceByKey().map().count() ? Deux : un seul shuffle, celui de reduceByKey.
3. Pourquoi préférer reduceByKey à groupByKey ? Il additionne d’abord sur chaque partition et n’envoie qu’un total par clé : environ 70 fois moins de shuffle au TP 2.
4. Votre join renvoie un résultat vide sans erreur : que vérifiez-vous ? Le type des clés, identique des deux côtés (int(c[0])).
5. Que change --master dans spark-submit ? L’endroit où s’exécutent les Executors. Le code métier ne change pas.
6. Un DataNode tombe : que se passe-t-il ? Les copies des blocs sur les autres DataNodes prennent le relais, et HDFS re-réplique.
7. RDD ou DataFrame : lequel est le plus rapide sur le top 10, et pourquoi ? Le DataFrame : Catalyst optimise le plan (broadcast join, projection poussée vers Parquet).
8. Pourquoi la division par zéro ne plante-t-elle pas sans le WHERE ? show() ne lit que les 20 premières lignes, aucune n’a une distance nulle. Spark est paresseux jusqu’au bout.