🚀 Apache Spark – Moteur Big Data (RDD, DataFrame, Spark SQL)

Qu'est-ce qu'Apache Spark ?

Apache Spark est un moteur de calcul (compute engine) open-source, conçu pour le traitement de données distribué (Big Data).

Une application Spark (ex: un spark-submit) s'exécute comme un ensemble de processus indépendants sur un cluster.

(Votre Script .py)
       │
       │ (1. Soumission : spark-submit)
       ▼
[ CLUSTER MANAGER ] (ex: YARN, Kubernetes)
 (Le "RH" : Négocie les ressources)
       │
       │ (2. Lance le Driver)
       ▼
+---------------------+
| [ DRIVER (JVM) ]    | (Le "Cerveau")
| (Votre 'main()')    |
| (SparkContext)      |
| (Catalyst Optimizer)|
| (Planifie le DAG)   |
+---------------------+
       │
       │ (3. Négocie les Executors)
       │
       └────────► [ CLUSTER MANAGER ]
                        │
                        │ (4. Lance les Executors)
                        │
    ┌───────────────────┴───────────────────┐
    ▼                                       ▼
+----------------+                      +----------------+
| [ EXECUTOR 1 ] | (Le "Muscle")        | [ EXECUTOR 2 ] |
| (JVM/Worker)   |                      | (JVM/Worker)   |
| (Exécute Task 1) |                      | (Exécute Task 2) |
| (Cache Data A) |                      | (Cache Data B) |
+----------------+                      +----------------+

• Cluster Manager (Gestionnaire) : (ex: YARN, Kubernetes). Le "gestionnaire de ressources". Spark lui "demande" des conteneurs (CPU/RAM).
• Driver (Pilote) : (1 par App) Le "cerveau". C'est le processus main() de votre application. Il planifie le travail (le DAG) et coordonne les Executors.
• Executor (Exécuteur) : (N par App) Les "muscles" (workers). Lancés par le Cluster Manager. Ils exécutent les Tâches (Tasks) (le calcul) et stockent (cache) les partitions de données en RAM.

Le RDD est l'API "bas niveau" originale de Spark. C'est une collection d'objets (immuable, distribuée) sur laquelle on peut opérer en parallèle.

Resilient : Si une partition (Executor) est perdue, Spark peut la recalculer (grâce au DAG).

Distribué : Le RDD est "coupé" en Partitions, stockées sur les Executors.

Dataset : Une collection de données (ex: lignes d'un log, objets Python).

RDD (Non-Structuré)

Le RDD est "agnostique" (il ne connaît pas le schéma). C'est une "boîte noire" d'objets (ex: RDD[String], RDD[MaClassePerso]).

Exemple (PySpark)

# 1. Créer un RDD (depuis un fichier)
lines_rdd = sc.textFile("s3://.../logs.txt")
# (RDD de Strings (lignes))

# 2. Transformation (map) - Lazy
errors_rdd = lines_rdd.filter(lambda line: "ERROR" in line)

# 3. Transformation (map) - Lazy
msgs_rdd = errors_rdd.map(lambda line: line.split(" ")[-1])

# 4. Action (Trigger) - Exécute le Job
results = msgs_rdd.collect()

Le DataFrame (DF) est l'API "haut niveau" moderne (introduite dans Spark SQL). C'est le standard de facto aujourd'hui.

DataFrame (Structuré)

Un DataFrame est un RDD (distribué) + un Schéma (Schema).

C'est une table (structurée) avec des colonnes nommées (similaire à un DataFrame Pandas ou une table SQL).

Exemple (PySpark)

# 1. Créer un DataFrame (lit le Parquet/Delta)
# (Le schéma est lu automatiquement)
df = spark.read.format("parquet").load("s3://.../users.parquet")
# df.printSchema()
# |-- user_id: long
# |-- age: integer
# |-- city: string

# 2. Transformation (select) - Lazy
# (On manipule des *colonnes*, pas des "lignes")
df_filtered = df.select("user_id", "age") \
                .where(df["age"] > 30)

# 3. Action (Trigger)
df_filtered.show()

Dataset (Statique - Scala/Java)

Un Dataset (Scala/Java) est un DataFrame "typé statiquement". (Ex: Dataset[Person]). C'est le "meilleur des deux mondes" (Structuré (DF) + Typé (RDD)). (L'API Python (PySpark) n'a pas de Datasets, elle n'a que des DataFrames).

Règle : Toujours utiliser les DataFrames (Spark SQL), sauf cas très spécifique (données binaires non-structurées).

Le Rôle de Catalyst (L'Optimiseur)

La raison principale est la performance, grâce à l'Optimiseur Catalyst (3.2).

Spark SQL est le module (et l'API) qui permet d'exécuter des requêtes SQL (Standard) sur des DataFrames (ou des sources de données).

Usage (PySpark)

Les APIs DataFrame et SQL sont unifiées (elles partagent le même moteur Catalyst).

(Session Spark 'spark' est créée)

# 1. Charger un DataFrame
df = spark.read.parquet("s3://.../users.parquet")

# 2. Enregistrer le DF comme une "Table" (temporaire, en mémoire)
df.createOrReplaceTempView("users_view")

# 3. (Magie) Requêter la "Vue" en SQL
# (Ceci est exécuté par le moteur Spark, pas par une BDD SQL)
df_results = spark.sql("""
    SELECT
        city,
        AVG(age) as avg_age
    FROM
        users_view
    WHERE
        age > 20
    GROUP BY
        city
    ORDER BY
        avg_age DESC
""")

df_results.show()

Usage : Permet aux Data Analysts (qui connaissent SQL) de faire du Big Data, sans apprendre PySpark/Scala.

Catalyst est le moteur d'optimisation (query optimizer) de Spark SQL. C'est le "cerveau" qui rend les DataFrames (2.2) et Spark SQL (3.1) rapides.

Flux (Simplifié)

Quand vous exécutez df.show() ou spark.sql() :

1. Code (DF/SQL) : (ex: df.select(...).filter(...)).
2. Catalyst (Analyse) : Convertit le code en Plan Logique (Unresolved).
3. Catalyst (Analyse) : Vérifie le "Catalogue" (Métadonnées) -> Plan Logique (Analyzed).
4. Catalyst (Optimisation Logique) : Applique des règles (ex: Predicate Pushdown (Pousser les filtres), Constant Folding). -> Plan Logique (Optimisé).
5. Catalyst (Optimisation Physique) : Génère plusieurs Plans Physiques (ex: "Hash Join" vs "Broadcast Join") et choisit le meilleur (basé sur les coûts).
6. Génération de Code (Codegen) : (Étape finale) Génère du bytecode Java (très rapide) pour ce plan.
7. Exécution : (Envoie le code (Tâches) aux Executors).

Spark (via spark.read et df.write) peut se connecter à (presque) n'importe quelle source de données.

spark.read.format(...).load(...)

C'est le concept fondamental de l'exécution Spark.

Spark divise les opérations en deux types : Transformations (Lazy) et Actions (Eager).

# (1. Transformation) -> Lazy. (0 sec)
# (Spark construit le DAG : "Etape 1: Lire")
df = spark.read.parquet("s3://...")

# (2. Transformation) -> Lazy. (0 sec)
# (Spark ajoute au DAG : "Etape 2: Filtrer")
df_filtered = df.where(col("age") > 30)

# (3. Action) -> Eager. (Déclenche le Job)
# (Maintenant, Spark exécute (1) et (2))
df_filtered.show()

Lorsqu'une "Action" (4.1) est appelée, le Driver (Catalyst) convertit le "Plan Logique" en DAG (Directed Acyclic Graph) d'exécution physique.

[ JOB 1 (ex: .show()) ]
   │
   ├─► [ STAGE 1 (Narrow) ] (ex: read + filter)
   │     ├─ Task 1 (Partition 1)
   │     ├─ Task 2 (Partition 2)
   │     └─ ... (Task N)
   │
   │ (--- SHUFFLE (4.3) ---)
   │
   └─► [ STAGE 2 (Wide) ] (ex: groupBy)
         ├─ Task 1 (Reducer 1)
         └─ Task 2 (Reducer 2)

• Job : (1 Job = 1 Action). (ex: df.write(...)).
• Stage (Étape) : Un "Stage" est un ensemble de Tâches qui peuvent être exécutées en parallèle (Narrow Transformation) sans Shuffle (4.3). Un Shuffle (ex: groupBy) coupe le DAG et crée un nouveau Stage.
• Task (Tâche) : (L'unité de travail) 1 Tâche = 1 "Partition" de données, exécutée sur 1 "Core" (CPU) d'un "Executor".

Le Shuffle (Mélange) est l'opération (physique) de redistribution des données sur le réseau (entre les Executors). C'est l'opération la plus coûteuse (lente) dans Spark.

Transformations "Narrow" vs "Wide"

• Narrow (Étroit) : (Rapide) Pas de Shuffle. Les données restent sur leur partition. (Ex: map, filter, select).
• Wide (Large) : (Lent) Nécessite un Shuffle. Les données (clés) doivent être déplacées entre les Executors. (Ex: groupByKey, reduceByKey, join (sauf Broadcast), repartition).

Exemple (groupByKey)

Pour compter les "Clés" (A, B, C), qui sont réparties sur 4 Executors :

1. (Stage 1) Executor 1 (A, B), Executor 2 (A, C), Executor 3 (B, C)...
2. (Shuffle - I/O Disque/Réseau) Executor 1 doit envoyer ses "B" et "C" aux autres. Executor 2 doit envoyer... (etc.).
3. (Stage 2) (Après Shuffle) Executor 1 a (A, A), Executor 2 a (B, B), Executor 3 a (C, C).
4. (Stage 2) Les Executors comptent (réduisent) leurs clés locales.

Optimisation : Toujours préférer reduceByKey (qui pré-agrège sur le "Mapper" (Stage 1)) à groupByKey (qui envoie tout (brut) au Shuffle).

L'option --deploy-mode (dans spark-submit) est cruciale. Elle définit où (physiquement) le processus Driver (1.2) va s'exécuter.

(Laptop) -> spark-submit (Job 1)
(Laptop) -> (Le terminal se libère)

(Sur le Cluster YARN)
  -> [Node 1 (Master)] -> (Driver (Job 1))
  -> [Node 2] -> (Executor 1)
  -> [Node 3] -> (Executor 2)

La Spark UI est l'interface web (GUI) de diagnostic de Spark. C'est l'outil n°1 pour déboguer la performance.

Elle est (par défaut) hébergée sur le Driver (1.2), sur le Port 4040 (http://[IP_Driver]:4040).

Onglets Clés (Diagnostic)

DStreams (Legacy) : L'ancien "Spark Streaming" (avant 2.x) était basé sur des RDDs (DStreams). (Obsolète).

Structured Streaming (Moderne)

Le Structured Streaming (Spark 2.x+) est le moteur de streaming moderne. Il est basé sur l'API DataFrame (2.2).

Le "DataFrame Infini"

Concept : Vous traitez un flux (Stream) de données (ex: Kafka) comme s'il s'agissait d'une table (DataFrame) statique qui "grandit" (append-only).

Vous écrivez une requête (Batch) une seule fois (ex: df.groupBy("type").count()). Spark (Moteur) se charge de l'exécuter en continu (micro-batch).

Exemple (PySpark)

# 1. Lecture (Source : Socket)
# (Crée un DF 'lines' qui représente le flux)
lines_df = spark.readStream \
    .format("socket") \
    .option("host", "localhost") \
    .option("port", 9999) \
    .load()

# 2. Transformation (DataFrame standard)
words_df = lines_df.select(
    explode(split(lines_df.value, " ")).alias("word")
)
word_counts_df = words_df.groupBy("word").count()

# 3. Écriture (Sink : Console)
# (Action 'writeStream' qui déclenche le Job)
query = word_counts_df.writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()

Kafka (le Bus de Messages) est la Source (Input) et le Sink (Output) le plus courant pour Structured Streaming.

df = spark.readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "host1:port1,...") \
  .option("subscribe", "mon_topic") \
  .load()

# (Le DF Kafka a des colonnes 'key', 'value' (binaire),
#  'topic', 'partition', 'timestamp')
df_string = df.selectExpr("CAST(value AS STRING)")

# (df_results est le résultat du 'groupBy')
# (Doit avoir une colonne 'value', ou 'key'/'value')

query = df_results.writeStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "host1:port1,...") \
  .option("topic", "topic_resultats") \
  .option("checkpointLocation", "/path/to/checkpoints") \
  .start()

Problème (Streaming) : Vous faites un groupBy(window("10 minutes")).count(). Que faire si un événement de 10h02 arrive en retard (à 10h15) (à cause du réseau) ?

Solution : Watermarking (Filigrane)

Le Watermarking (withWatermark) dit à Spark : "J'accepte les données en retard (ex: 10 minutes de retard), mais au-delà, ferme la fenêtre (agrégation) et ignore-les."

C'est la façon dont Spark gère le compromis Latence vs Complétude.

df_logs = ... (avec "timestamp")

df_window = df_logs \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window("timestamp", "5 minutes") # (Agrège par fenêtre de 5 min)
    ) \
    .count()

Résultat : Spark attendra 10 minutes (le "watermark") avant de finaliser (écrire) le compte de la fenêtre "10:00-10:05".

PySpark est l'API Python pour Spark. C'est le langage dominant en Data Science et Data Engineering (sur Spark).

Architecture (Py4J)

Problème : Spark (Driver/Executor) tourne sur la JVM (Java/Scala). Python (CPython) tourne dans un processus séparé.

Solution : Py4J (la "glue").

[ Processus Python (PySpark) ]
  (ex: df.select("age"))
  │
  │ (Socket Py4J)
  │
  ▼
[ Processus JVM (Spark Driver) ]
  │
  │ (Catalyst Optimizer)
  │ (Crée le Plan)
  ▼
[ Executors (JVM) ]

Overhead : Les opérations DataFrame (3.1) sont rapides (le code Python est "traduit" en plan Spark). Les opérations RDD (rdd.map(lambda...)) sont lentes (Spark doit sérialiser le code Python, l'envoyer aux Executors, lancer un "Worker Python", exécuter, et resérialiser les résultats -> Coûteux).

MLlib est la bibliothèque (intégrée) de Machine Learning distribué de Spark. (Basée sur l'API DataFrame).

Concept : Pipeline (ML)

MLlib (comme Scikit-learn) est basé sur le concept de Pipelines (chaînes d'étapes).

(Exemple PySpark : Pipeline de Classification)
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler
from pyspark.ml.classification import LogisticRegression

# 1. Étapes de "Featurization" (Pré-traitement)
# (Transformer "ville" (string) en "ville_idx" (nombre))
indexer = StringIndexer(inputCol="ville", outputCol="ville_idx")

# (Assembler les colonnes "features" en 1 vecteur)
assembler = VectorAssembler(inputCols=["age", "ville_idx"],
                            outputCol="features")

# 2. Étape de "Modèle" (Entraînement)
lr = LogisticRegression(featuresCol="features", labelCol="a_achete")

# 3. Définir le Pipeline
pipeline = Pipeline(stages=[indexer, assembler, lr])

# 4. Entraîner (Fit)
model = pipeline.fit(training_data_df)

# 5. Prédire
predictions_df = model.transform(test_data_df)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, sum, count, when, lit

# 1. Créer la Session (Driver)
spark = SparkSession.builder \
    .appName("MonApp") \
    .getOrCreate()

# 2. Lire (Source)
df = spark.read.format("parquet").load("/path/to/data")
# df = spark.read.csv("/path/to/csv", header=True, inferSchema=True)

# --- 3. TRANSFORMATIONS (Lazy) ---

# Sélectionner des colonnes
df_select = df.select("col_A", "col_B")
df_select = df.select(col("col_A"), (col("col_B") * 2).alias("col_B_double"))

# Filtrer (Where)
df_filter = df.where(col("age") > 30)
df_filter = df.filter((col("age") > 30) & (col("ville") == "Paris"))

# Ajouter/Modifier une colonne (withColumn)
df_with = df.withColumn("age_plus_10", col("age") + 10)
df_with = df.withColumn("categorie",
    when(col("age") < 18, "Mineur")
    .when(col("age") < 65, "Adulte")
    .otherwise("Senior")
)
df_with = df.withColumn("constante", lit(123)) # Ajouter une constante

# Renommer / Supprimer
df_rename = df.withColumnRenamed("nom", "nouveau_nom")
df_drop = df.drop("col_inutile_1", "col_inutile_2")

# Jointure (Join)
df_joined = df_A.join(
    df_B,
    df_A["user_id"] == df_B["user_id"], # Clé de Join
    "inner" # (Type: inner, left, right, full)
)

# Agrégation (GroupBy)
df_agg = df.groupBy("ville") \
           .agg(
                count("*").alias("nb_users"),
                avg("age").alias("age_moyen")
           )

# --- 4. ACTIONS (Eager) ---

# Afficher (Debug)
df_agg.show(10)

# Afficher le Schéma (Debug)
df_agg.printSchema()

# Compter (Action)
count = df_agg.count()

# Collecter (Action - DANGER : RAM Driver)
# (Ramène TOUTES les données (résultat) au Driver)
results_list = df_agg.collect()

# Écrire (Sink)
df_agg.write \
    .format("delta") \
    .mode("overwrite") \
    .save("/path/to/output")