Partie 2 – Spark Core : RDD, PairRDD et exécution

• 10h45 – 12h00
• Apports : 25 min
• TP 2 : 30 min
• Partage + quiz express : 20 min

Et si on ouvrait le capot ? Au TP 1, votre count() a produit 1 job et N tâches. Pourquoi N ?

Le RDD : une collection découpée en partitions

Le RDD (Resilient Distributed Dataset) est la structure de base de Spark. Un fichier trips.csv lu par Spark devient un RDD découpé en partitions, et chaque partition est confiée à un Executor sur une machine du cluster.

RDD = trips.csv
┌────────────┬────────────┬────────────┬────────────┐
│ Partition 1│ Partition 2│ Partition 3│ Partition 4│
└─────┬──────┴─────┬──────┴─────┬──────┴─────┬──────┘
      └── Machine A – Executor ──┘  └── Machine B – Executor ──┘

Lettre	Signification	Conséquence
Resilient	Si une partition est perdue (machine en panne), Spark la reconstruit grâce à sa lignée – la « recette » des transformations qui l’ont produite	Pas de copie des données, juste la recette
Distributed	Chaque partition est traitée par un Executor, en parallèle	1 partition = 1 tâche
Dataset	Collection immuable : transformer un RDD, c’est en créer un nouveau	Pas de modification en place

Analogie de la lignée : si on perd une page du livre, on refait la recette plutôt que de stocker une photocopie de chaque page.

Un RDD est immuable : comment traite-t-on les données si on ne peut jamais les modifier ?

Une fois créé, le contenu d’un RDD ne peut jamais être modifié en place. Pour traiter les données, chaque étape applique une transformation qui crée un nouveau RDD en mémoire à partir du précédent, tout en gardant un pointeur vers son parent (la lignée) :

  RDD initial             RDD 1                   RDD 2                   RDD 3
  (en mémoire)         (en mémoire)            (en mémoire)            (en mémoire)
 ┌────────────┐       ┌────────────┐          ┌────────────┐          ┌────────────┐
 │ Partition 1│ ───▶  │ Partition 1│  ───▶    │ Partition 1│  ───▶    │ Partition 1│
 ├────────────┤       ├────────────┤          ├────────────┤          ├────────────┤
 │ Partition 2│ ───▶  │ Partition 2│  ───▶    │ Partition 2│  ───▶    │ Partition 2│
 ├────────────┤       ├────────────┤          ├────────────┤          ├────────────┤
 │ Partition 3│ ───▶  │ Partition 3│  ───▶    │ Partition 3│  ───▶    │ Partition 3│
 └────────────┘       └────────────┘          └────────────┘          └────────────┘
              transf. 1              transf. 2               transf. 3
              (filter)                 (map)                 (filter)

Définition — Transformation : opération de création d’un nouveau RDD à partir d’un RDD existant, sans modifier le RDD d’origine.

🔍 Vulgarisation : Pourquoi l'immuabilité est-elle indispensable sur un cluster distribué ?

On pourrait se demander : « Pourquoi ne pas modifier directement les données en place dans la RAM pour économiser de la mémoire ? »

Imaginez un calcul distribué sur 10 machines qui ajoute 10 % aux tarifs des courses :

• Si la mémoire était modifiable en place (mutable) et que la machine n°4 plante à la moitié de sa partition : certaines lignes de la partition 4 ont déjà +10 %, d’autres non. Personne ne sait où elle s’est arrêtée ! Il faudrait tout annuler sur les 10 machines et recharger depuis le disque.
• Grâce à l’immuabilité (RDD0 → RDD1 → RDD2) : le RDD0 d’origine n’a jamais été touché. Si la machine n°4 tombe en panne pendant le calcul de RDD2, Spark consulte simplement la lignée (lineage : RDD2 = RDD0.filter(...).map(...)) et rejoue cette recette uniquement pour la partition 4 sur une autre machine, pendant que les 9 autres continuent sans s’arrêter.

Le PairRDD : l’outil des agrégations

Un PairRDD est un RDD de couples (clé, valeur). C’est lui qui permet reduceByKey, join, sortByKey, countByKey. L’exemple ci-dessous est exactement le TP 2 : clé = zone de départ, valeur = montant.

(132, 45.2)  (132, 12.0)  (48, 30.1)  (48, 9.5)
          │ reduceByKey(lambda a, b: a + b)
          ▼
(132, 57.2)  (48, 39.6)        = chiffre d'affaires par zone

reduceByKey pré-agrège sur chaque partition avant d’échanger les données entre machines ; groupByKey envoie toutes les valeurs sur le réseau. Le bonus du TP 2 le mesure : ≈ 480 Ko de Shuffle Write pour groupByKey contre ≈ 7 Ko pour reduceByKey, soit environ 70 fois moins de données échangées.

🧮 Trace pas à pas : Pourquoi reduceByKey envoie 70× moins de données sur le réseau que groupByKey ?

Supposons que la Partition 1 (Machine A) contienne 1 000 courses parties de l’aéroport JFK (zone 132) :

❌ Avec groupByKey() : aucun calcul local avant le réseau !
   Machine A (Partition 1) envoie les 1 000 couples un par un sur le réseau :
   (132, 45.2), (132, 12.0), (132, 70.0), (132, 55.4) … [× 1 000 lignes envoyées sur le câble !]

✅ Avec reduceByKey(lambda a, b: a + b) : addition locale (Map-side combine) AVANT le réseau !
   Machine A additionne d'abord ses 1 000 courses en mémoire locale :
   45.2 + 12.0 + 70.0 + 55.4 + … = 48 520.0
   Puis elle n'envoie qu'UN SEUL couple sur le réseau :
   (132, 48520.0)   [1 seule ligne au lieu de 1 000 !]

Sur notre fichier de 120 000 courses réparties sur 265 zones :

• groupByKey transfère les 120 000 montants sur le réseau (≈ 480 Ko).
• reduceByKey n’envoie au maximum que 265 sous-totaux par partition (≈ 7 Ko), soit ≈ 70× moins de trafic réseau. Sur 100 Go en production, c’est la différence entre un job qui finit en 30 secondes et un job qui plante avec OutOfMemoryError.

Transformations, actions et évaluation paresseuse

Définitions : Transformation vs Action

Dans Spark, toutes les opérations sur un RDD se divisent en deux familles :

• Transformation (Mémoire) : opération qui crée un nouveau RDD à partir d’un autre RDD en appliquant un traitement sur les données initiales (map, filter, flatMap, reduceByKey…).
  • Tant qu’on ne demande pas de résultat final, Spark ne calcule rien : il note simplement la recette en mémoire (un pointeur du nouveau RDD vers son RDD parent).
  • Règle d’or : Création d’un RDD sans sortir de résultat ➔ Transformation.
• Action (Disque ou Affichage écran) : opération qui s’applique sur un RDD pour sortir un résultat concret. Les données calculées sont rassemblées et sont :
  • soit affichées à l’écran (rapatriées vers le Driver : count(), collect(), take(10)),
  • soit stockées dans un fichier sur disque ou sur une autre machine (saveAsTextFile(), write.parquet()).
  • Exemples concrets de résultats d’action : afficher le nombre de personnes vaccinées par jour, le nombre de personnes infectées par jour, ou sauvegarder le chiffre d’affaires des taxis par zone (notre TP 2).

                        transformations (map, filter, reduceByKey…)
                        ─────────── en mémoire (RAM) ───────────▶  Nouveau RDD
                       │                                           (immuable)
       RDD initial ────┤  ◀──── pointeur vers le parent (lignée)
       (immuable)      │
                       │ actions (count, reduce, collect, saveAsTextFile…)
                       ▼
             [ Disque / Écran ]
   Sauvegarder (fichier) ou Afficher (Driver)

Piège classique : reduceByKey (Transformation) vs reduce (Action)

Ne vous fiez pas uniquement au mot « Reduce » :

• reduceByKey(func) est une TRANSFORMATION : elle additionne ou combine les valeurs par clé (par exemple par jour ou par zone) et renvoie un nouveau PairRDD distribué en mémoire. Aucun résultat n’est encore sorti !
• reduce(func) est une ACTION : elle combine tous les éléments du RDD en une seule valeur finale (un nombre, un total global) et la renvoie immédiatement au Driver.

Exemple sur notre fil rouge : 4 transformations + 1 action

lignes  = sc.textFile("data/trips.csv")                          # transformation (RDD 0)
donnees = lignes.filter(lambda l: not l.startswith("VendorID"))   # transformation (RDD 1)
champs  = donnees.map(lambda l: l.split(","))                    # transformation (RDD 2)
longs   = champs.filter(lambda c: float(c[4]) > 10)              # transformation (RDD 3 : trajets > 10 miles)
longs.count()                                                    # ACTION : le calcul démarre ici !

Les quatre premières lignes répondent instantanément (quelques millisecondes) : rien n’est lu sur le disque, rien n’est calculé. Spark construit un plan, le DAG. C’est l’action (count, collect, take, saveAsTextFile…) qui déclenche l’exécution. C’est l’évaluation paresseuse (lazy evaluation) : elle permet à Spark d’optimiser l’ensemble du plan avant de lancer quoi que ce soit.

Combien de lectures et d’écritures sur disque ? (MapReduce vs Spark)

Comparons ce qui se passe concrètement sur le disque entre l’approche classique (MapReduce / étapes sur disque) et Spark lorsque l’on enchaîne 2 opérations, 4 itérations, 10 itérations ou 20 itérations :

Approche sur disque (MapReduce : chaque étape lit et réécrit sur disque) :
  Disque ──▶ [Lecture 1] ──▶ Étape 1 ──▶ [Écriture disque 1]
         ──▶ [Lecture 2] ──▶ Étape 2 ──▶ [Écriture disque 2]
         ──▶ [Lecture 3] ──▶ Étape 3 ──▶ [Écriture disque 3] …

Approche Spark (pipelining + mémoire RAM) :
  Disque ──▶ [1 seule lecture] ──▶ RAM : RDD0 ─▶ RDD1 ─▶ RDD2 ─▶ RDD3 ──▶ Résultat / 1 écriture

1. Sur nos 4 opérations ci-dessus (textFile → filter → map → filter → count) :
   • Spark fusionne (pipeline) les transformations dans un seul stage en mémoire : 1 seule lecture disque et 0 écriture disque intermédiaire.
   • Une ligne d’en-tête éliminée au premier filter n’est même pas découpée par split(",").
2. Sur un traitement répétitif ou itératif (avec .cache() en mémoire) :
   • Dans MapReduce, chaque itération doit tout relire sur HDFS au début et tout réécrire sur HDFS à la fin ($N$ itérations = $N$ lectures + $N$ écritures disque = $2N$ passages sur disque).
   • Dans Spark avec .cache(), la 1re action lit le disque et garde les partitions du RDD en RAM ; toutes les itérations suivantes lisent directement la RAM (10 à 50 Go/s) au lieu du disque (0,1 à 0,5 Go/s) :

Nombre d’opérations / itérations (ex. sur 100 Go)	MapReduce (disque à chaque tour)	Spark sans .cache()	Spark avec .cache() (en RAM)	Gain Spark en mémoire
2 opérations enchaînées	2 lectures + 2 écritures (4 passages disque = 400 Go)	1 lecture (pipelining en RAM)	1 lecture disque + RAM	2× à 3× plus rapide
4 itérations (ex. 4 étapes ou 4 tours de K-Means)	4 lectures + 4 écritures (8 passages disque = 800 Go)	4 lectures disque (recalcul de la lignée)	1 lecture disque + 3 lectures en RAM (+ 1 écriture finale)	6 passages disque évités (~10× à 20× plus rapide)
10 itérations (ex. PageRank, validation croisée)	10 lectures + 10 écritures (20 passages disque = 2 To)	10 lectures disque	1 lecture disque + 9 lectures en RAM (+ 1 écriture finale)	18 passages disque évités (~50× plus rapide)
20 itérations (ex. régression logistique / ML — benchmark de Berkeley)	20 lectures + 20 écritures (40 passages disque = 4 To)	20 lectures disque	1 lecture disque + 19 lectures en RAM (+ 1 écriture finale)	38 passages disque évités → jusqu’à 100× plus rapide (≈ 0,8 s/tour en RAM contre ≈ 80 s/tour sur disque)

Transformations (paresseuses — en mémoire)	Actions (déclenchent un job — disque ou écran)
map, filter, flatMap, distinct, sample	count, collect, first, take(n)
reduceByKey, groupByKey, join, sortBy	takeOrdered(n), reduce, saveAsTextFile

Narrow ou wide : où est le shuffle ?

Type	Exemples	Ce qui se passe	Coût
Narrow	map, filter	Chaque partition de sortie dépend d’une seule partition d’entrée : traitement sur place, sans réseau	Faible
Wide	reduceByKey, join, groupByKey, sortBy	Chaque partition de sortie reçoit des morceaux de toutes les partitions d’entrée (par clé) : les données changent de machine = shuffle	Élevé : réseau et disque

Message clé : 1 shuffle = 1 nouveau stage. C’est le premier levier de performance d’un job Spark.

🔍 Vulgarisation : Narrow vs Wide et le Shuffle (l'analogie du tri de cartes postales)

Imaginez 4 employés assis chacun à une table séparée, avec une pile de 30 000 cartes postales (4 partitions = 120 000 cartes) :

1. Transformation Narrow (filter, map) :
   • Consigne : « Gardez uniquement les cartes écrites en bleu et tamponnez-les. »
   • Chaque employé travaille seul à sa table, sans parler aux autres. C’est instantané, 0 échange entre les tables.
2. Transformation Wide (reduceByKey, join, groupByKey) :
   • Consigne : « Calculez le total des timbres par ville de destination. »
   • L’employé n°1 a des cartes pour Paris, Lyon et Marseille ; l’employé n°2 en a aussi pour Paris, Lyon et Marseille !
   • Ils sont obligés de s’arrêter, de trier leurs cartes par ville, et de se lever pour s’échanger les paquets entre les tables afin que l’employé n°1 reçoive toutes les cartes de Paris et l’employé n°2 toutes celles de Lyon.
   • Cet échange physique entre toutes les tables, c’est le Shuffle : Spark doit écrire des fichiers temporaires sur disque et les transférer sur le réseau avant de pouvoir commencer le stage suivant.

SparkContext ou SparkSession ?

SparkContext (Spark 1.x)	SparkSession (depuis Spark 2.0)
Point d’entrée historique de Spark Core	Point d’entrée unique : SQL, DataFrames, streaming
Crée les RDD : sc.textFile, sc.parallelize	Contient le SparkContext : spark.sparkContext
Un seul par application	Remplace SQLContext et HiveContext
Complété à l’époque par SQLContext et HiveContext	Seul point d’entrée en Spark Connect (pas de RDD côté client)

On crée toujours une SparkSession ; sc = spark.sparkContext si on a besoin des RDD. Avec Spark Connect (client léger qui parle à un serveur Spark distant), l’API RDD n’existe pas côté client : une raison de plus de préférer les DataFrames (partie 4).

RDD vs DataFrame vs Dataset : quelle différence ?

Un RDD manipule des lignes brutes ou des objets sans savoir quelles « colonnes » ils contiennent. Un DataFrame organise les mêmes données sous forme de table avec des colonnes nommées et typées (comme une table SQL), ce qui permet à Spark d’optimiser automatiquement les calculs :

      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     │
 Spark voit de simples chaînes de texte :          └────────┴──────────┴──────────┘
 à vous de faire split(";") et c[0], c[1]…         Spark connaît le nom et le type de chaque colonne.

Voici le comparatif complet des trois structures de données de Spark :

Critère (Feature)	RDD	DataFrame	Dataset
Immuable (Immutable)	Oui	Oui	Oui
Tolérant aux pannes (Fault tolerant)	Oui (via la lignée)	Oui	Oui
Typage fort à la compilation (Type-safe)	Oui	Non (types vérifiés à l’exécution)	Oui (Scala / Java uniquement)
Schéma de colonnes (Schema)	Non (données brutes)	Oui (colonnes nommées et typées)	Oui
Optimiseur d’exécution (Catalyst / Tungsten)	Non (exécute votre code tel quel)	Oui (réorganise et accélère le plan)	Oui
Niveau d’abstraction (Level)	Bas (Low-level : comment faire)	Haut (High-level : quoi faire)	Haut (High-level)

Pourquoi Dataset n’existe-t-il pas en PySpark ? Parce que Python est un langage à typage dynamique (sans compilateur de types comme Scala ou Java). En PySpark, on utilise donc les DataFrames (qui sont techniquement des Dataset[Row] sur la JVM) dans 95 % des cas, et les RDD lorsqu’on traite des données non structurées (texte libre, binaire) ou qu’on veut comprendre le fonctionnement interne de Spark (comme dans le TP 2).

💡 Sous le capot : Pourquoi Python est lent sur les RDD mais aussi rapide que Scala sur les DataFrames ?

Spark est écrit en Scala et tourne dans une machine virtuelle Java (JVM) :

• Quand vous utilisez un RDD en Python (rdd.map(lambda x: ...)) :
  • La JVM ne comprend pas le code bytecode de votre lambda Python !
  • L’Executor JVM est obligé de sérialiser chaque ligne (via pickle), de l’envoyer par un tuyau local (socket) à un processus Python externe, qui exécute la lambda, puis renvoie le résultat à la JVM : JVM Executor ──(pickle)──▶ Processus Python ──(pickle)──▶ JVM Executor
  • Cet aller-retour permanent coûte cher sur des millions de lignes.
• Quand vous utilisez un DataFrame ou Spark SQL (df.filter(F.col("trip_distance") > 10)) :
  • Python n’exécute aucune ligne de donnée ! Il envoie juste l’expression logique "trip_distance > 10" à la JVM.
  • C’est l’optimiseur Catalyst et le moteur Tungsten qui compilent et exécutent tout à 100 % dans la JVM, exactement à la même vitesse qu’un programme écrit en Scala.

Du code au cluster : DAG, schedulers et workers

 RDD Objects        DAG Scheduler            Task Scheduler            Worker
 votre code     ──▶ découpe le DAG en   ──▶  envoie les tâches    ──▶  les Executors
 construit le DAG   stages aux shuffles      via le cluster manager    exécutent les tâches

• DAG (Directed Acyclic Graph) : le plan d’exécution, sans boucle.
• 1 action = 1 job → plusieurs stages (un par shuffle, plus un) → 1 tâche par partition.
• Les tâches d’un même stage tournent en parallèle.

Dans la Spark UI du TP 2 (onglet Jobs › DAG Visualization), vous verrez exactement ceci : un stage 1 avec textFile, filter, map, puis une flèche (le shuffle de reduceByKey), puis un stage 2 avec l’agrégation et takeOrdered. Compter les tâches du premier stage = compter les partitions : c’est la réponse à l’accroche de la partie.

À vous de jouer

→ TP 2 – Top 10 des zones avec les RDD (30 min)

Puis le quiz express : transformation ou action ? Combien de stages ?

Partage et remédiation

• Pourquoi reduceByKey est-il plus rapide que groupByKey ? Il pré-agrège dans chaque partition avant le shuffle : environ 70 fois moins de données échangées (mesuré : ≈ 7 Ko contre ≈ 480 Ko).
• Combien de stages avez-vous vus, et pourquoi ? Deux : la lecture + map, puis l’agrégation après le shuffle. Le join et le sortBy de l’étape 5 en ajoutent d’autres.
• À retenir : 1 shuffle = 1 nouveau stage. C’est là que se joue la performance.

Transition : « Les données sont encore sur un seul disque. Et si elles ne tenaient plus ? » → Partie 3 – Hadoop, HDFS et Spark sur un cluster