Dockerisation d’un cluster Spark

from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.feature import StringIndexer
from pyspark.ml import Pipeline
from pyspark.ml.regression import DecisionTreeRegressor

# Créer une session Spark
spark = SparkSession.builder.appName("MyML").getOrCreate()

# Charger le fichier CSV des vols dans un DataFrame Spark
psdf_flights = spark.read.csv(
    "file:///opt/bitnami/spark/data/flights.csv", header=True, inferSchema=True
)

# Remplir les valeurs manquantes avec 0
psdf_flights = psdf_flights.fillna(0)

# Utiliser StringIndexer pour numériser les colonnes "ORIGIN_AIRPORT", "DESTINATION_AIRPORT" et "AIRLINE"
indexer_origin = StringIndexer(
    inputCol="ORIGIN_AIRPORT", outputCol="ORIGIN_AIRPORT_Index"
)
indexer_destination = StringIndexer(
    inputCol="DESTINATION_AIRPORT", outputCol="DESTINATION_AIRPORT_Index"
)
indexer_airline = StringIndexer(inputCol="AIRLINE", outputCol="AIRLINE_Index")

# Créer un pipeline pour appliquer les transformations en une seule étape
pipeline = Pipeline(stages=[indexer_origin, indexer_destination, indexer_airline])

# Appliquer le pipeline de transformation sur les données de vol
psdf_indexe = pipeline.fit(psdf_flights).transform(psdf_flights)

# Utiliser VectorAssembler pour assembler toutes les colonnes de caractéristiques en une seule colonne de vecteur
featureAssembler = VectorAssembler(
    inputCols=[
        "YEAR", "MONTH", "DAY", "DAY_OF_WEEK", "AIRLINE_Index", "FLIGHT_NUMBER",
        "ORIGIN_AIRPORT_Index", "DESTINATION_AIRPORT_Index", "SCHEDULED_DEPARTURE",
        "DEPARTURE_TIME", "DEPARTURE_DELAY", "TAXI_OUT", "WHEELS_OFF", "SCHEDULED_TIME",
        "ELAPSED_TIME", "AIR_TIME", "DISTANCE", "WHEELS_ON", "TAXI_IN", "SCHEDULED_ARRIVAL",
        "ARRIVAL_TIME", "DIVERTED", "CANCELLED", "AIR_SYSTEM_DELAY", "SECURITY_DELAY",
        "AIRLINE_DELAY", "LATE_AIRCRAFT_DELAY", "WEATHER_DELAY"
    ],
    outputCol="IndependantFeatures",
) # Cette ligne crée une nouvelle colonne avec les caractéristiques assemblées

# Appliquer la transformation pour assembler les caractéristiques
output = featureAssembler.transform(psdf_indexe)

# Sélectionner les caractéristiques indépendantes et la variable cible (ARRIVAL_DELAY)
spdf_to_ML = output.select("IndependantFeatures", "ARRIVAL_DELAY")

# Diviser les données en ensembles d'entraînement (75%) et de test (25%) avec une graine pour la reproductibilité
train_data_dt, test_data_dt = spdf_to_ML.randomSplit([0.75, 0.25], seed=41)

# Initialiser le régressseur d'arbre de décision
regressor_dt = DecisionTreeRegressor(
    featuresCol="IndependantFeatures", labelCol="ARRIVAL_DELAY", maxBins=800
)

# Entraîner le modèle sur les données d'entraînement
model_dt = regressor_dt.fit(train_data_dt)

# Faire des prédictions sur l'ensemble de test
psdf_pred_results_dt = model_dt.transform(test_data_dt)

# Imprimer un message indiquant la fin du traitement
print("FIN DU TRAITEMENT")

# Sauvegarder le modèle entraîné dans le répertoire spécifié
model_dt.write().overwrite().save("/opt/bitnami/spark/models/model_dt")

# Arrêter la session Spark
spark.stop()