⚡ Apache Spark – Le Moteur du Big Data

Spark a été créé pour résoudre les limitations de Hadoop MapReduce (MR).

Pour le développement, on n'a pas besoin d'un cluster. pip install pyspark lance Spark en **"Local Mode"** (le Driver et l'Executor sont sur votre machine).

# 1. (Recommandé) Créer un environnement virtuel
python -m venv spark_env
source spark_env/bin/activate

# 2. Installer PySpark (API Python pour Spark)
pip install pyspark

# 3. (Optionnel: pour Jupyter) Trouver Spark
pip install findspark

Lancement (Script ou Notebook)

import findspark
findspark.init()

import pyspark
from pyspark.sql import SparkSession

# (Le point d'entrée, voir 2.1)
spark = SparkSession.builder.appName("MonAppLocal").getOrCreate()

print(spark.version)

Spark est un système **Driver/Executors** (Maître/Esclaves).

• Driver (Pilote) : Le "cerveau" (là où SparkSession tourne). Il planifie le travail (le DAG).
• Cluster Manager : Le "RH" (YARN, Kubernetes). Négocie les ressources (machines).
• Executors (Exécuteurs) : Les "bras" (Workers). Exécutent les "Tasks" (calculs) sur les partitions de données.

+-------------------------+
| Votre App (PySpark)     |
| (Contient le "Driver")  |
| (Planifie le DAG)       |
+-------------------------+
      | (1. Demande des ressources)
      ▼
+-------------------------+
| Cluster Manager         |
| (YARN / Kubernetes)     |
+-------------------------+
      | (2. Alloue les Executors)
      ▼
+-------------------------+  +-------------------------+
| Executor 1 (Worker)     |  | Executor 2 (Worker)     |
| (Cache) (Task 1, Task 3)|  | (Cache) (Task 2, Task 4)|
+-------------------------+  +-------------------------+
      | (3. Lit/Écrit)             | (3. Lit/Écrit)
      ▼                          ▼
+----------------------------------------+
| Stockage Distribué (S3, HDFS, Parquet) |
+----------------------------------------+

Depuis Spark 2.0, le SparkSession (conventionnellement appelé spark) est le point d'entrée unique pour toutes les fonctionnalités de Spark (SQL, DataFrame, Streaming).

C'est lui qui configure et lance le "Driver".

from pyspark.sql import SparkSession

# 1. Créer (ou récupérer) la session
spark = SparkSession.builder \
    .appName("MonAnalyseDeVentes") \
    .master("local[4]") \ // ("local[4]" = 4 cœurs en mode local)
    .config("spark.some.config.option", "valeur") \
    .getOrCreate()

# 2. Utiliser 'spark' pour lire/écrire/SQL
df = spark.read.csv("data.csv")
spark.sql(...)

# 3. Accéder au SparkContext (API bas niveau)
sc = spark.sparkContext

Spark a deux API de données :

Le Catalyst Optimizer (moteur d'optimisation de Spark) ne comprend que les DataFrames. Il réorganise les opérations (ex: "pousser" un filter avant un join) pour la performance.

C'est le concept le plus important. Spark est **LAZY (Paresseux)**.

Quand vous appliquez une Transformation (ex: .filter()), Spark n'exécute **rien**. Il construit un "plan d'exécution", un **DAG (Directed Acyclic Graph)**.

Le calcul (le "Job") n'est déclenché que lorsqu'une Action (ex: .show(), .count()) est appelée.

Diagramme (Lazy Evaluation)

// (1. Transformation 1: Build DAG)
df = spark.read.parquet("ventes.parquet")
// (Plan: Lire 'ventes.parquet')

// (2. Transformation 2: Update DAG)
df_fr = df.filter(df.pays == "FR")
// (Plan: Lire -> Filtre)

// (3. Transformation 3: Update DAG)
df_agg = df_fr.groupBy("produit").sum("prix")
// (Plan: Lire -> Filtre -> GroupBy)

// (RIEN NE S'EST ENCORE PASSÉ)

// (4. ACTION: Déclenche l'exécution du DAG)
df_agg.show()
// (Spark lance le Job, les Executors tournent)

L'API spark.read et df.write.

Parquet est le format N°1 du Big Data (colonnaire, compressé, optimisé pour Spark).

Lecture (spark.read)

// 1. Parquet (Format préféré)
df = spark.read.parquet("s3a://mon-bucket/data/ventes/")

// 2. CSV (Moins performant)
df = spark.read.csv(
    "data.csv",
    header=True,
    inferSchema=True // (Lent, évitez en prod, voir 3.2)
)

// 3. (Générique)
df = spark.read.format("jdbc").option(...).load()

Écriture (df.write)

// 1. Parquet (avec partitionnement)
df.write \
  .mode("overwrite") \ // ('overwrite', 'append', 'error')
  .partitionBy("annee", "mois") \ // (Crée des dossiers ex: /annee=2024/)
  .parquet("s3a://mon-bucket/output/ventes_clean/")
  
// 2. CSV
df.write.mode("append").csv("output.csv")

Laisser Spark "deviner" le schéma (inferSchema=True) est lent (Spark doit lire le fichier 2 fois). En production, on définit **toujours** le schéma.

Exemple (StructType)

from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType
)

# 1. Définir le schéma
schema = StructType([
    StructField("id_client", IntegerType(), nullable=False),
    StructField("nom", StringType(), nullable=True),
    StructField("prix_total", DoubleType(), nullable=True)
])

# 2. Utiliser le schéma à la lecture
df = spark.read.csv(
    "data.csv",
    header=True,
    schema=schema // (Rapide, pas d'inférence)
)

# 3. Afficher le schéma
df.printSchema()

L'API DataFrame ressemble beaucoup à Pandas, mais les opérations sont **Lazy**.

from pyspark.sql.functions import F

# (df = DataFrame chargé)

# 1. Sélection (Select)
# (Retourne un nouveau DF)
df_select = df.select("col1", "col2")

# (Syntaxe 'dot')
df_select_dot = df.select(df.col1, df.col2)

# 2. Filtre (Filter / Where)
df_filter = df.filter(df.age > 30)
df_filter_multi = df.filter(
    (df.age > 30) & (df.ville == "Paris")
)

# 3. Ajouter/Modifier Colonne (withColumn)
df_new_col = df.withColumn(
    "age_plus_un", // (Nom nouvelle colonne)
    df.age + 1     // (Logique)
)

# (Utiliser F.* pour les fonctions SQL)
df_new_col_F = df.withColumn(
    "nom_upper",
    F.upper(df.nom)
)

# 4. Renommer
df_renamed = df.withColumnRenamed("nom", "nom_client")

# 5. Supprimer
df_dropped = df.drop("col_inutile")

Similaire à Pandas groupby (Split-Apply-Combine). C'est une **Wide Transformation** (implique un "Shuffle" coûteux des données sur le cluster).

On utilise pyspark.sql.functions (alias F) pour les fonctions d'agrégation.

from pyspark.sql.functions import F

// 1. Syntaxe simple
df_agg_simple = df.groupBy("ville").mean("salaire")
// (Renomme 'avg(salaire)')

// 2. Syntaxe .agg() (Préférée, multi-aggs)
df_agg_multi = df.groupBy("ville").agg(
    F.mean("salaire").alias("salaire_moyen"),
    F.sum("ventes").alias("ventes_totales"),
    F.count("id_client").alias("nb_clients")
)

// 3. GroupBy Multi-colonnes
df_agg_multi_cols = df.groupBy("ville", "departement").agg(
    F.max("age").alias("age_max")
)

// (Ceci est LAZY. Lancer .show() pour exécuter)
df_agg_multi_cols.show()

Spark SQL permet d'exécuter des requêtes SQL (ANSI-SQL) directement sur les DataFrames. C'est souvent plus simple que l'API Python pour des JOIN ou GROUP BY complexes.

# (df = DataFrame chargé)

# 1. Enregistrer le DataFrame comme "Vue" temporaire
df.createOrReplaceTempView("ventes")
df_clients.createOrReplaceTempView("clients")

# 2. Exécuter du SQL (via spark.sql)
# (Ceci est une Transformation LAZY)
df_sql = spark.sql("""
    SELECT 
        c.nom,
        SUM(v.prix) as total_depense
    FROM ventes v
    JOIN clients c ON v.client_id = c.client_id
    WHERE v.pays = 'FR'
    GROUP BY c.nom
    ORDER BY total_depense DESC
""")

# 3. Déclenche le calcul
df_sql.show()

Structured Streaming est l'API haut niveau (basée sur les DataFrames) pour le streaming (ex: lire de Kafka, fichiers S3 entrants...).

Spark traite le flux (stream) comme un "DataFrame infini" (unbounded table). Il exécute des "micro-batchs" (ex: toutes les 30s) sur les nouvelles données.

# 1. Lecture (readStream) (ex: depuis Kafka)
df_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "host:port") \
    .option("subscribe", "mon_topic") \
    .load()

# 2. Transformations (identiques à l'API DF/SQL)
df_processed = df_stream.selectExpr("CAST(value AS STRING)") \
    .filter(...)

# 3. Écriture (writeStream) (Déclenche)
query = df_processed.writeStream \
    .outputMode("append") \ // ('append', 'complete', 'update')
    .format("parquet") \ // (Sink: Parquet, Kafka, Console)
    .option("path", "/output/data") \
    .option("checkpointLocation", "/checkpoints") \ // (Crucial pour la tolérance)
    .trigger(processingTime='1 minute') \
    .start()

query.awaitTermination()

MLlib est la bibliothèque ML de Spark (similaire à Scikit-learn, mais distribuée).

VectorAssembler : Étape N°1 obligatoire. MLlib requiert que toutes les features (X) soient dans une seule colonne "vecteur".

Pipeline : Comme sklearn, on utilise des Pipelines pour chaîner (Scaler, OHE, VectorAssembler, Modèle).

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import LogisticRegression

# 1. Assembler les features en 1 vecteur
assembler = VectorAssembler(
    inputCols=["age", "salaire", "ohe_ville"],
    outputCol="features"
)

# 2. Scaler
scaler = StandardScaler(inputCol="features", outputCol="scaledFeatures")

# 3. Modèle
lr = LogisticRegression(featuresCol="scaledFeatures", labelCol="target")

# 4. Pipeline
pipeline = Pipeline(stages=[assembler, scaler, lr])

# 5. Entraîner
model = pipeline.fit(train_data)

# 6. Prédire
predictions = model.transform(test_data)

La syntaxe est similaire, mais la philosophie est radicalement différente.

Pandas API on Spark (pyspark.pandas) : Un effort pour implémenter l'API Pandas "Eager" au-dessus du moteur Spark "Lazy". Utile pour la transition.

Le Cluster Manager est le "RH" (voir 1.4) qui alloue les Executors (machines).

Spark est le standard de facto pour l'ETL (Extract-Transform-Load) et le ML à grande échelle (Big Data).

Ressources pour apprendre et travailler avec Apache Spark.