Mise en place d'un labo Apache Spark local avec des exemples pratiques : guide d'implémentation

Introduction

Un labo Spark local permet d'apprendre vite, de tester des idées en sécurité et de construire des habitudes reproductibles avant d'utiliser des systèmes partagés. Dans ce guide, vous allez :

Installer Spark et vérifier le mode local.
Créer un répertoire de labo propre avec données, logs et checkpoints.
Exécuter des exemples batch et streaming que vous pouvez inspecter.
Soumettre des jobs de manière programmatique et ajouter un test simple.

À la fin, vous disposerez d'un flux de travail réutilisable pour le dépannage, les expériences et le développement (Apache Spark development). Ce guide sert d'Apache Spark setup pragmatique, avec de vrais Apache Spark examples, pour faciliter l'Apache Spark testing dans un Apache Spark local lab.

Vue d'ensemble du flux

Adoptez un flux simple et répétable qui garde les changements petits et visibles :

Définir un objectif étroit et des critères de succès.
Préparer les dossiers du labo et des données d'échantillon.
Configurer les valeurs par défaut de Spark pour des exécutions et logs cohérents.
Exécuter du code en mode local et inspecter l'UI Spark.
Capturer les résultats dans des dossiers de sortie versionnés.
Ajouter un petit test pour protéger les transformations clés.
Itérer par petites étapes, en changeant une chose à la fois.

Cette séparation des étapes réduit le rework et facilite l'isolement des problèmes.

Configuration du labo local

Pré-requis

CPU : 4 cœurs
RAM : 8 Go minimum
Disque : 10 Go libres
Java 11 ou 17
Python 3.9+ si vous utilisez PySpark

Créer les dossiers du labo

mkdir -p ~/labs/spark/data/batch/input
mkdir -p ~/labs/spark/data/batch/output
mkdir -p ~/labs/spark/data/stream/input
mkdir -p ~/labs/spark/logs/events
mkdir -p ~/labs/spark/checkpoints

Installer Spark (binaires précompilés)

Choisissez une distribution Spark 3.5.x précompilée pour Hadoop 3.
Décompressez-la, par exemple dans ~/tools/spark-3.5.1-bin-hadoop3.
Définissez les variables d'environnement :

macOS ou Linux (bash/zsh) :

export SPARK_HOME=~/tools/spark-3.5.1-bin-hadoop3
export PATH="$SPARK_HOME/bin:$PATH"
# Définissez JAVA_HOME vers votre JDK installé
# Exemple macOS :
export JAVA_HOME=$(/usr/libexec/java_home -v 17)

Windows (PowerShell, chemins variables) :

setx SPARK_HOME "C:\\tools\\spark-3.5.1-bin-hadoop3"
setx PATH "%SPARK_HOME%\\bin;%PATH%"
# Définissez JAVA_HOME vers le JDK installé (utilisateur ou système)

Configurer les valeurs par défaut de Spark

Copiez le modèle :

cp "$SPARK_HOME/conf/spark-defaults.conf.template" "$SPARK_HOME/conf/spark-defaults.conf"

Éditez spark-defaults.conf et ajoutez :

spark.eventLog.enabled true
spark.eventLog.dir file:///home/youruser/labs/spark/logs/events
spark.sql.shuffle.partitions 4

Ajustez le chemin des journaux d'événements selon votre OS et votre utilisateur.

Vérifier l'installation

spark-shell --version
pyspark --version

Lancez une session interactive et quittez :

pyspark --master local[2]
# Dans le shell :
spark.range(5).show()
# Visitez http://localhost:4040 pour voir l'UI Spark pendant l'activité.
exit()

Exemples pratiques

1) Agrégation batch : CSV vers Parquet

Préparez un petit CSV :

cat > ~/labs/spark/data/batch/input/customers.csv << 'EOF'
customer_id, name, country, spend
1, Alice, US,120.5
2, Bob, DE,95
3, Cara, US,250
4, Dan, FR,40
5, Eve, US,
EOF

Créez batch_example.py :

# batch_example.py
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.functions import col
import argparse

def transform(df):
    # Filtrer les spend nuls et typer la colonne
    clean = (
        df.filter(col("spend").isNotNull())
          .withColumn("spend_usd", col("spend").cast("double"))
    )
    # Agrégation par pays avec moyenne arrondie
    agg = (
        clean.groupBy("country")
             .agg(
                 F.count("*").alias("customers"),
                 F.round(F.avg("spend_usd"), 2).alias("avg_spend")
             )
             .orderBy("country")
    )
    return agg

def main(input_path: str, output_path: str):
    spark = SparkSession.builder.appName("batch-agg").getOrCreate()
    df = (spark.read
               .option("header", "true")
               .option("inferSchema", "true")
               .csv(input_path))

    result = transform(df)
    (result.write
           .mode("overwrite")
           .parquet(output_path))

    result.show(truncate=False)
    spark.stop()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    main(args.input, args.output)

Exécutez avec spark-submit :

cd ~/labs/spark
spark-submit \
  --master local[4] \
  --name batch-agg \
  batch_example.py \
  --input data/batch/input \
  --output data/batch/output/parquet

Inspectez les sorties :

ls -R data/batch/output/parquet

Pendant l'exécution, ouvrez l'UI :

http://localhost:4040 pour les DAGs et stages
Consultez les logs d'événements dans ~/labs/spark/logs/events

2) Structured Streaming : source fichier vers console

Créez stream_example.py :

# stream_example.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col, count

spark = (
    SparkSession.builder
    .appName("word-count-stream")
    .master("local[2]")
    .getOrCreate()
)

schema = "ts timestamp, word string"
stream = (
    spark.readStream
         .schema(schema)
         .json("data/stream/input")
)

counts = (
    stream.withWatermark("ts", "1 minute")
          .groupBy(window(col("ts"), "30 seconds"), col("word"))
          .agg(count("*").alias("n"))
)

query = (
    counts.writeStream
          .outputMode("append")
          .format("console")
          .option("truncate", "false")
          .trigger(processingTime="5 seconds")
          .start()
)

print("Streaming started. Append JSON lines to data/stream/input ...")
query.awaitTermination()

Démarrez le stream :

cd ~/labs/spark
python3 stream_example.py

Dans un autre terminal, ajoutez des événements :

cd ~/labs/spark
cat > data/stream/input/batch1.json << 'EOF'
{"ts":"2024-01-01T10:00:00","word":"spark"}
{"ts":"2024-01-01T10:00:10","word":"spark"}
{"ts":"2024-01-01T10:00:20","word":"lab"}
EOF

Vous verrez des comptes glissants dans la console. Visitez http://localhost:4040 pour observer les requêtes de streaming et l'état pendant l'exécution.

3) Test minimal pour la transformation batch

Ajoutez test_batch_example.py :

# test_batch_example.py
import pytest
from pyspark.sql import SparkSession, Row
from batch_example import transform

@pytest.fixture(scope="session")
def spark():
    return (
        SparkSession.builder
        .appName("test-batch-agg")
        .master("local[2]")
        .getOrCreate()
    )

def test_transform_basic(spark):
    rows = [
        Row(customer_id=1, name="A", country="US", spend=100.0),
        Row(customer_id=2, name="B", country="US", spend=50.0),
        Row(customer_id=3, name="C", country="DE", spend=70.0),
        Row(customer_id=4, name="D", country="US", spend=None),
    ]
    df = spark.createDataFrame(rows)
    out = transform(df)
    got = { (r["country"], r["customers"], float(r["avg_spend"])) for r in out.collect() }
    assert ("DE", 1, 70.0) in got
    assert ("US", 2, 75.0) in got

Exécutez le test :

pytest -q

Plan pilote local

Choisissez une tâche mesurable et rapidement vérifiable :

Objectif : convertir un petit CSV en Parquet et calculer des moyennes par pays.
Périmètre : <= 10 Mo de données d'entrée, 1 dossier de sortie, 1 fonction de transformation.
Critères de succès :
Le Parquet de sortie existe avec les colonnes et le nombre de lignes attendus.
L'UI Spark montre des stages terminés sans échec.
Temps d'exécution < 30 secondes sur votre machine.
Timebox : 1 à 2 heures, incluant l'installation et un test basique.

Pourquoi cela fonctionne : c'est étroit, mesurable et facile à inspecter en local. Le feedback est rapide et vous pourrez grandir vers d'autres sources plus tard (par exemple, ajouter Kafka ou HDFS une fois le chemin local solide). Vous pourrez ensuite orchestrer avec Apache Airflow ou NiFi.

Dépannage et conseils

Erreurs Java : vérifiez que java -version correspond au JDK attendu (11 ou 17). Définissez JAVA_HOME explicitement.
pyspark introuvable : assurez-vous que SPARK_HOME/bin est dans votre PATH.
UI Spark indisponible : l'UI n'est active que lorsqu'une application tourne. Démarrez un shell ou un job, puis visitez http://localhost:4040.
Conflits de port : si le port 4040 est pris, Spark essaie 4041, 4042, etc. Lisez la sortie du terminal pour connaître le port effectif.
Mémoire : en local, essayez --driver-memory 2g et réglez spark.sql.shuffle.partitions à une petite valeur (par exemple 4).
Permissions de fichiers : gardez votre labo sous votre répertoire personnel pour éviter les problèmes de droits.
Reproductibilité : fixez la version de Spark, épinglez les versions Python et gardez les données d'exemple sous contrôle de version.

Sujets connexes à explorer ensuite :

Kafka : basculez la source de fichiers vers une source Kafka quand un broker est disponible.
HDFS : orientez entrées et sorties vers HDFS pour des tests plus volumineux.
Apache Airflow ou NiFi : orchestrez les jobs une fois le pilote local stabilisé.

Conclusion

Vous avez maintenant un Apache Spark local lab sûr, deux exemples pratiques (batch et streaming), un test minimal et un plan pilote clair. Gardez des itérations petites, vérifiez les sorties et les performances via l'UI et les logs, puis ajoutez seulement ensuite des intégrations comme Kafka, HDFS, Apache Airflow ou NiFi. Cette approche maintient l'apprentissage rapide et des échecs peu coûteux, tout en construisant pas à pas vers des charges de production. Elle soutient une démarche d'Apache Spark development rigoureuse, avec un Apache Spark setup reproductible et des Apache Spark examples utiles pour l'Apache Spark testing au quotidien.