Configuration d'un laboratoire Apache Spark local avec exemples pratiques

Intro

Un laboratoire Spark local permet d'apprendre rapidement, de tester des idées en toute sécurité et de bâtir des habitudes reproductibles avant de toucher aux systèmes partagés. Dans ce guide vous allez :

Installer Spark et vérifier le mode local.
Créer une arborescence de laboratoire propre avec données, journaux et points de contrôle.
Exécuter des exemples batch et streaming que vous pouvez inspecter.
Soumettre des jobs programmatiquement et ajouter un test minimal.

À la fin, vous disposerez d'un workflow réutilisable pour le dépannage, l'expérimentation et le développement.

Workflow Overview

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

Définissez un objectif précis et des critères de succès.
Préparez les dossiers du laboratoire et les données d'exemple.
Configurez les valeurs par défaut de Spark pour des exécutions et des logs cohérents.
Lancez le code en mode local et inspectez l'interface Spark UI.
Capturez les résultats dans des dossiers de sortie versionnés.
Ajoutez un petit test pour protéger les transformations critiques.
Itérez par petites étapes, en changeant une seule chose à la fois.

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

Local Lab Setup

Prerequisites

CPU : 4 cœurs
RAM : 8 Go minimum
Disque : 10 Go libres
Java 11 ou 17 (vérifiez avec java -version)
Python 3.9+ si vous utilisez PySpark

Create the lab folders

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

Install Spark (pre‑built binaries)

Téléchargez Spark 3.5.1 pré‑construit pour Hadoop 3 depuis l'archive officielle.
Décompressez‑le, par exemple dans ~/tools/spark-3.5.1-bin-hadoop3.
Définissez les variables d'environnement.

macOS / Linux (bash/zsh)

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

Windows (PowerShell)

setx SPARK_HOME "C:\tools\spark-3.5.1-bin-hadoop3"
setx PATH "%SPARK_HOME%\bin;%PATH%"
# Définissez JAVA_HOME sur le chemin de votre JDK comme variable d'environnement utilisateur ou système

Configure Spark defaults

Copiez le modèle et éditez le fichier de configuration :

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

Ajoutez les lignes suivantes à spark-defaults.conf (remplacez youruser par votre nom d'utilisateur réel) :

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

Verify the install

spark-shell --version
pyspark --version

Démarrez une session PySpark interactive et faites un test rapide :

pyspark --master local[2]
# Dans le shell :
spark.range(5).show()
# Visitez http://localhost:4040 pour voir la Spark UI tant que le shell est actif.
exit()

Practical Examples

1) Batch aggregation: CSV to Parquet

Préparez un petit CSV avec une valeur manquante à la dernière ligne :

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
from pyspark.sql import functions as F
from pyspark.sql.functions import col
import argparse

def transform(df):
    clean = (
        df.filter(col("spend\)).isNotNull())
          .withColumn("spend_usd", col("spend\)).cast("double\)
    )
    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‑le 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 la sortie :

ls -R data/batch/output/parquet

Pendant l'exécution, ouvrez la Spark UI à http://localhost:4040 pour visualiser les DAG et les étapes, et consultez les journaux d'événements dans ~/labs/spark/logs/events.

2) Structured Streaming: file source to console

Créez stream_example.py :

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

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

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

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

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

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

Démarrez le job streaming :

cd ~/labs/spark
python3 stream_example.py

Dans un autre terminal, ajoutez quelques é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 devriez voir les comptages défiler dans la console. Rendez‑vous sur http://localhost:4040 pour observer les requêtes streaming et les opérations d'état pendant l'exécution.

3) Minimal test for the batch transform

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 (\n        SparkSession.builder\n        .appName("test-batch-agg\)\n        .master("local[2]\n        .getOrCreate()\n    )

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

Lancez le test :

pytest -q

Local Pilot Plan

Choisissez une tâche que vous pouvez mesurer et inspecter rapidement :

Goal : Convertir un petit CSV en Parquet et calculer les moyennes par pays.
Scope : ≤ 10 Mo de données d'entrée, un dossier de sortie, une fonction de transformation.
Success criteria :
Le Parquet de sortie existe avec les colonnes attendues (country, customers, avg_spend) et un nombre de lignes égal au nombre de pays distincts.
La Spark UI affiche des étapes terminées sans échec.
Temps d'exécution sous 30 secondes sur votre machine.
Timebox : 1 à 2 heures, incluant la configuration et un test de base.

Pourquoi ça fonctionne : c'est ciblé, mesurable et facile à inspecter localement. Vous obtenez un retour rapide et pouvez étendre plus tard vers d'autres sources (par exemple, ajouter Kafka ou HDFS une fois le chemin local solide).

Troubleshooting and Tips

Erreurs Java : Assurez‑vous que java -version correspond au JDK attendu (11 ou 17). Définissez JAVA_HOME explicitement.
pyspark introuvable : Vérifiez que $SPARK_HOME/bin est bien dans votre PATH.
Spark UI non disponible : L'UI ne tourne que pendant qu'une application Spark est active. Lancez un shell ou un job, puis visitez http://localhost:4040.
Conflits de port : Si le port 4040 est occupé, Spark essaiera 4041, 4042, etc. Regardez la sortie du terminal pour le port réel.
Mémoire : Pour des exécutions locales, essayez --driver-memory 2g --executor-memory 2g et gardez spark.sql.shuffle.partitions petit (par exemple 4).
Permissions de fichiers : Gardez votre laboratoire sous votre répertoire personnel pour éviter les problèmes de droits.
Reproductibilité : Fixez la version de Spark, épinglez les versions des packages Python (par exemple via requirements.txt), et conservez les données d'exemple dans le contrôle de version.

Sujets connexes à explorer ensuite :

Kafka : remplacez la source de fichiers streaming par une source Kafka dès que vous avez un broker.
HDFS : pointez les entrées et sorties vers HDFS pour des tests plus volumineux.
Apache Airflow ou NiFi : orchestrez les jobs une fois votre pilote local stable.

Conclusion

Vous disposez maintenant d'un laboratoire Spark local sûr, de deux exemples pratiques, d'un test minimal et d'un plan pilote clair. Procédez par petites itérations, vérifiez les sorties et les performances dans l'UI et les logs, et n'ajoutez des intégrations comme Kafka, HDFS, Airflow ou NiFi qu'une fois le socle local maîtrisé. Cette approche garde l'apprentissage rapide et les échecs peu coûteux tout en vous préparant au travail à l'échelle de la production.