Aujourd'hui, nous nous attaquons à un morceau conséquent de la programmation parallèle et distribuée, et nous vous proposons ce tutoriel pour comprendre l'implémentation de Spark en local et dans un cluster de machines. ## Une histoire de logement ![Photo de @naletu sur Unsplash](https://images.unsplash.com/photo-1512420502149-ebd993111c3d?q=80&w=1974&auto=format&fit=crop&ixlib=rb-4.0.3&ixid=M3wxMjA3fDB8MHxwaG90by1wYWdlfHx8fGVufDB8fHx8fA%3D%3D) Afin de déployer un code suffisamment gourmand pour mériter d'être "sparké", nous sommes partis sur un problème ne nécessitant pas nécessairement un calcul très complexe, mais plutôt l'obligation d'effectuer ces calculs sur de nombreuses lignes avec une consolidation importante des données. Nous constatons actuellement que l'opinion publique débat énormément, notamment sur les réseaux sociaux, à propos des logements dits vacants ou inoccupés, dans un contexte de crise du logement très prononcée. En effet, d'une part, les biens disponibles à la location et à la vente deviennent de plus en plus rares, et d'autre part, la demande pour occuper ces lieux ne cesse d'augmenter. Un article très récent de [Le Monde](https://www.lemonde.fr/societe/article/2023/12/05/paris-veut-s-attaquer-aux-logements-inoccupes-pour-stopper-la-baisse-de-la-population_6204014_3224.html) atteste qu'un logement sur cinq à Paris serait inoccupé, alors que la demande continue de grimper. La question est alors simple : Paris est-il un cas isolé ou un grain de semoule dans un problème plus grand qu'il n'y paraît ? Nous avons ainsi décidé d'essayer de déterminer la part de logements dits vacants ou inoccupés dans l'ensemble des communes de France, en la rapportant à la population plutôt qu'au nombre de logements ### Source et travail de la donnée Nous avons décidé de nous baser sur les sources et données suivantes : - [Les logements vacants par commune et par EPCI en 2020](https://www.data.gouv.fr/fr/datasets/logements-vacants-du-parc-prive-par-anciennete-de-vacance-par-commune-et-par-epci/#/resources) - [L'historique des populations communales (sélection des données de 2020)](https://www.insee.fr/fr/statistiques/3698339) Ces deux fichiers comportent une colonne "numéro de la commune" dont nous pouvons nous servir afin de joindre les deux tableaux en un. Nous supprimerons aussi dans les deux fichiers les lignes dites orphelines que nous n'avons pas pu associer à une autre ligne dans l'autre fichier. ## Libre accès de notre code L'ensemble de notre dépôt Git est accessible sur le [GitLab de l'université de Lille en public](https://gitlab.univ-lille.fr/valentin.bout.etu/ppd). Vous y trouverez notamment les éléments des sections suivantes, ainsi que notre IaaS construit avec Terraform et les scripts d'installation pour Hadoop, HDFS et YARN. Maintenant, place à la technique pure et dure, miam ;). ## Mise en place hors Spark. Pour commencer, nous allons écrire notre programme sans Spark afin d'étudier le temps d'exécution du programme sans Spark et mieux comprendre la logique de notre code. L'ensemble du code exprimé ci-dessous se retrouve dans notre dépôt [GitLab](https://gitlab.univ-lille.fr/valentin.bout.etu/ppd/-/tree/master/no_spark?ref_type=heads) dans le dossier *no_spark*. ### Dépendances nécessaires Pour faire fonctionner notre programme, nous avons besoin de pandas et numpy sur Python. Nous utilisons Python en version **3.11.2**. Commençons par mettre en place notre environnement Python : ```sh python -m venv env # Création d'un environnement pour notre projet source env/bin/activate # Activation de notre environnement pip install pandas numpy # Installation de pandas et numpy ``` Nous allons également créer deux dossiers. Le premier nous permettra de stocker notre code Python sans Spark. Dans le second, nous y mettrons nos données téléchargeables au lien communiqué précédemment. ```sh mkdir no_spark data ``` Nous avons désormais tous les éléments à notre disposition pour créer notre premier programme naïf sans Spark. ### Un premier code Le pseudocode de notre solution à notre problème peut s'exprimer de la manière suivante : ``` population = lire_fichier_csv() logements = lire_fichier_csv() pop_et_log = fusionner(population, logements, par='communes') pop_et_log[poucentage] = (nombre_logement_vacant * 100) / population_commune pop_et_log[poucentage-2] = (nombre_logement_vacant * 100) / nombre_total_logement sauvegarder_csv(pop_et_log) ``` En d'autres termes, nous devons importer les deux fichiers de données, fusionner ces deux fichiers en utilisant comme critère commun le numéro INSEE de la commune, et calculer les pourcentages du nombre de logements vacants par rapport à la population et par rapport au total de logements. Finalement, il ne nous reste plus qu'à sauvegarder ces données. ### Implémentation en python Nous allons transformer notre pseudo code précédent en code python. ```python # Importation des dépendances nécessaires import pandas as pd import numpy as np # Importation des deux fichiers dans notre dossier data base_pop = pd.read_csv('../data/base-pop-historiques-1876-2020.csv') logements = pd.read_csv('../data/logements-vacants-du-parc-prive-par-commune-au-01012021-lovac.csv') """ Convertis un objet en int si possible Si ele est None, 0 est retourné """ def convert_to_int(ele): nele = str(ele).replace(',', '') if ele is None or np.isnan(float(nele)): return 0 return int(nele) # On convertis les elements en entier base_pop['PMUN20'] = base_pop['PMUN20'].apply(int) logements['Nb_logvac_pp_010120'] = logements['Nb_logvac_pp_010120'].apply(convert_to_int) logements['Nb_log_pp_2020'] = logements['Nb_log_pp_2020'].apply(convert_to_int) # Effectuer une jointure pandas joined_df = base_pop.merge(logements, left_on='CODGEO', right_on='INSEE_COM', how='inner') # Calculer la colonne pourcentage joined_df['Pourcentage vacant par population'] = (joined_df['Nb_logvac_pp_010120'] * 100) / joined_df['PMUN20'] joined_df['Pourcentage vacant par logement'] = (joined_df['Nb_logvac_pp_010120'] * 100) / joined_df['Nb_log_pp_2020'] # Sélectionner uniquement les colonnes nécessaires result_df = joined_df[['CODGEO', 'LIBGEO', 'Nb_log_pp_2020', 'Nb_logvac_pp_010120', 'PMUN20', 'Pourcentage vacant par population', 'Pourcentage vacant par logement']] # Trions les resultats result_df.sort_values(by=['Pourcentage vacant par population', 'Pourcentage vacant par logement'], inplace=True, ascending=False) # Afficher les resultats dans la sortie standart print(result_df) result_df.to_csv('../data/nospark-export.csv') ``` ### Et le temps d'exécution ? Notre programme, grâce au très bon fonctionnement de Pandas, ne prend plus énormément de temps à s'exécuter (moins de 10 secondes en moyenne). Le but de ce tutoriel est essentiellement de démontrer la facilité d'utilisation et d'installation de Spark. Cependant, nous obtiendrons de meilleurs résultats si nous disposions d'un jeu de données beaucoup plus important que "seulement" des milliers de lignes de données. ### Quelques notes additionnelles Avant de passer à la mise en place de notre code avec Spark, nous souhaitions revenir sur un point durant le développement de notre premier programme. Nous avions tout d'abord privilégié une approche naïve consistant à parcourir l'ensemble de nos dataframes et les fusionner avec une simple double boucle for en Python dans notre code. Il est apparu que cette première version était impossible à réaliser malgré un ordinateur très puissant, nous sommes parvenus à être à court de RAM. Nous avons ainsi, dans un second temps, créé le programme que nous venons de partager avec l'utilisation de jointures entre deux dataframes, transformant radicalement le temps d'exécution et la faisabilité de notre problème. ## Mise en place de spark en local Notre code sans Spark fonctionnant parfaitement, il est maintenant temps de paralléliser l'exécution de notre code sur une machine locale en utilisant les différents cœurs mis à notre disposition par la machine. ### Installation de spark en local L'installation de Spark en local consiste simplement à l'installation de la bibliothèque, ici en Python. En reprenant les prérequis précédents, nous avons juste à changer les dépendances avec pip. ```sh pip install numpy pyspark[pandas_on_spark] pyspark # Installation de numpy pandas.pyspark et pyspark ``` Nous n'avons plus besoin d'installer pandas classique car nous allons utiliser pandas ré-écris par spark. ### Transformation de notre code Pour transformer notre code avec Spark, il suffit simplement de modifier les imports et de créer une session Spark. La bibliothèque PySpark Pandas ayant la même API que sa sœur Pandas, le changement de bibliothèque se fait sans encombre. Modifications des imports et création d'une session spark : ``` import numpy as np import pyspark.pandas as pd from pyspark.sql import SparkSession spark = SparkSession.builder.getOrCreate() # --- Le reste du code est exactement identique ``` Vous pouvez retrouver notre code transformé sur le [GitLab dans le fichier main.py](https://gitlab.univ-lille.fr/valentin.bout.etu/ppd/-/blob/master/main.py?ref_type=heads) ET VOILÀ, nous venons de transformer notre code sans Spark avec Spark. Ce code est maintenant prêt à être exécuté avec Python en utilisant Spark sur une machine ou sur un cluster de machines, et spoiler... c'est l'étape suivante. ### Le temps, toujours le temps L'exécution de notre programme avec Spark est en réalité plus lente que sans Spark. En effet, la mise en place de la bibliothèque et son temps (environ quelques secondes) ne sont pas un investissement très rentable pour de petits programmes comme le nôtre. Pour autant, avec un jeu de données beaucoup plus important, nous pourrions observer sans problème le gain de temps effectif que peut nous apporter Spark. ## Parallélisation sur un cluster de machine (Spark, Hadoop YARN et HDFS) Spark peut être utilisé en mode cluster pour exécuter des charges de travail réparties entre plusieurs nœuds. Plusieurs gestionnaires de cluster sont compatibles avec Spark, dont Hadoop YARN. Pour stocker nos données ainsi que le résultat de nos calculs, nous utilisons HDFS, un système de fichier distribué compatible avec Hadoop YARN. ### Installation de Hadoop YARN et HDFS Nous réalisons cette installation sur des instances Ubuntu 22.04 avec l'utilisateur `ubuntu`⁣ : La première étape de notre installation est d'initialiser 3 VMs : l'une sera un gestionnaire de nœuds (node-master) et les deux autres seront des nœuds de travail (node1, node2). Sur nos 3 VMs, nous installons les dépendances nécessaires et nous téléchargeons Hadoop YARN depuis le repo officiel : ```bash sudo apt update sudo apt install -y openjdk-8-jre-headless echo 'export JAVA_HOME="/usr/lib/jvm/java-8-openjdk-amd64"' >> ~/.profile echo 'JAVA_HOME="/usr/lib/jvm/java-8-openjdk-amd64"' >> /etc/environment HADOOP_VERSION="3.5.0" wget https://dlcdn.apache.org/hadoop/common/hadoop-$HADOOP_VERSION/hadoop-$HADOOP_VERSION.tar.gz tar -xzf hadoop-$HADOOP_VERSION.tar.gz mv hadoop-$HADOOP_VERSION.tar.gz hadoop echo 'export PATH="${PATH}:/home/ubuntu/hadoop/bin:/home/ubuntu/hadoop/sbin" export HADOOP_CONF_DIR=/home/ubuntu/hadoop/etc/hadoop export LD_LIBRARY_PATH="/home/ubuntu/hadoop/lib/native:$LD_LIBRARY_PATH"' >> ~/.bashrc source ~/.bashrc ``` Avant de démarrer nos nœuds Hadoop YARN, nous devons modifier certains fichiers de configuration pour indiquer l'adresse de nos nœuds de travail, paramétrer HDFS et configurer YARN. `~/hadoop/etc/hadoop/code-site.xml` : ```xml fs.default.name hdfs://node-master:9000 ``` \ `~/hadoop/etc/hadoop/hdfs-site.xml` : ```xml dfs.namenode.name.dir /home/ubuntu/hadoop/data/nameNode dfs.datanode.data.dir /home/ubuntu/hadoop/data/dataNode dfs.replication 1 ``` \ `~/hadoop/etc/hadoop/hdfs-site.xml` : ```xml mapreduce.framework.name yarn yarn.app.mapreduce.am.env HADOOP_MAPRED_HOME=/home/ubuntu/hadoop mapreduce.map.env HADOOP_MAPRED_HOME=/home/ubuntu/hadoop mapreduce.reduce.env HADOOP_MAPRED_HOME=/home/ubuntu/hadoop yarn.app.mapreduce.am.resource.mb 512 mapreduce.map.memory.mb 256 mapreduce.reduce.memory.mb 256 ``` \ `~/hadoop/etc/hadoop/yarn-site.xml` : ```xml yarn.acl.enable 0 yarn.resourcemanager.hostname node-master yarn.nodemanager.aux-services mapreduce_shuffle yarn.nodemanager.resource.memory-mb 1536 yarn.scheduler.maximum-allocation-mb 1536 yarn.scheduler.minimum-allocation-mb 128 yarn.nodemanager.vmem-check-enabled false ``` \ `~/hadoop/etc/hadoop/workers`: ``` node1 node2 ``` Nous pouvons maintenant lancer nos nœuds : ```bash # Initialisation de HDFS hdfs namenode -format start-dfs.sh # Lancement de YARN start-yarn.sh ``` ### Installation de Spark Notre cluster est maintenant prêt à recevoir des charges de travail à exécuter. Nous devons donc installer Spark afin de les créer, puis de les envoyer au cluster. Nous téléchargeons Spark à partir du référentiel officiel. ```bash SPARK="3.5.0" wget https://dlcdn.apache.org/spark/spark-$HADOOP_VERSION/spark-$HADOOP_VERSION-bin-hadoop3.tgz tar -xvf spark-$HADOOP_VERSION-bin-hadoop3.tgz mv spark-$HADOOP_VERSION-bin-hadoop3 spark echo "export PATH=/home/ubuntu/spark/bin:$PATH export SPARK_HOME=/home/ubuntu/spark" >> ~/.bashrc source .bashrc mv $SPARK_HOME/conf/spark-defaults.conf.template $SPARK_HOME/conf/spark-defaults.conf echo "spark.master yarn spark.driver.memory 512m spark.yarn.am.memory 512m spark.executor.memory 512m spark.eventLog.enabled true spark.eventLog.dir hdfs://node-master:9000/spark-logs spark.history.provider org.apache.spark.deploy.history.FsHistoryProvider spark.history.fs.logDirectory hdfs://node-master:9000/spark-logs spark.history.fs.update.interval 10s spark.history.ui.port 18080" >> $SPARK_HOME/conf/spark-defaults.conf # Création du dossier de logs Spark sur HDFS hdfs dfs -mkdir /spark-logs ``` ### Installation de dépendances Python et des données de notre script Notre charge de travail est un script Python nécessitant les bibliothèques numpy et pandas. Le plus simple est de procéder à leur installation sur chaque nœud. L'utilisation de sudo permet d'installer les bibliothèques pour tous. ```bash sudo python3 -m pip install pandas numpy pyspark[pandas_on_spark] pyspark ``` Notre script nécessite les données de deux fichiers CSV. Pour rendre ces données accessibles lors de l'exécution sur le cluster Spark, nous devons transférer ces fichiers sur HDFS. ```bash hdfs dfs -mkdir data hdfs dfs -put data/logements-vacants-du-parc-prive-par-commune-au-01012021-lovac.csv data/base-pop-historiques-1876-2020.csv data ``` ### Lancement de notre charge de travail sur le cluster Le lancement de notre charge de travail se fait grâce à la commande suivante : ```bash spark-submit --master yarn --deploy-mode cluster --driver-memory 1g --executor-memory 1g --executor-cores 1 main.py ``` Une fois de charges de travail terminée, les données résultantes du script sont disponibles sur HDFS dans le dossier `data/exports` : ```bash hdfs dfs -ls data/exports hdfs dfs -get 'data/exports/*.csv' . ``` ## Et notre problème de logement dans tout cela ? Bien que notre problématique n'était qu'un prétexte pour tester Spark, il est tout de même important d'apporter une réponse : Paris n'est pas un cas isolé dans ce problème de logement vacant. Notamment, on y retrouve le village de Saint-Véran dans les Alpes ou la station Montclar ou Orcières. En d'autres termes, le problème de logement vacant semble encore plus accentué dans les zones touristiques et favorisé pendant les vacances, très probablement, car il doit s'agir de résidences secondaires.