Sommaire

Spark.MLlib

MLlib is Spark’s machine learning (ML) library. Its goal is to make practical machine learning scalable and easy. At a high level, it provides tools such as:

  • ML Algorithms: common learning algorithms such as classification, regression, clustering, and collaborative filtering
  • Featurization: feature extraction, transformation, dimensionality reduction, and selection
  • Pipelines: tools for constructing, evaluating, and tuning ML Pipelines
  • Persistence: saving and load algorithms, models, and Pipelines
  • Utilities: linear algebra, statistics, data handling…

Préparation à la manip

Créez un sous-répertoire MLlib dans le répertoire pyspark:

cd pyspark
mkdir MLlib
cd MLlib

Ce sera le répertoire de travail pour cet énoncé.

Dans un second Terminal, déplacez-vous dans le répertoire de votre machine où vous avez téléchargé le dépôt git lors du TP #1, et mettez-le à jour à l’aide de la commande

git pull

Vous serez amené à transférer ces fichiers vers le Namenode avec la commande docker cp ... (vous savez comment faire maintenant !).

Prise en main

Quelques informations

Trois catégories usuelles d’apprentissage artificiel sont : Classification, Clustering et Collaborative Filtering.

  • Classification: Gmail uses a machine learning technique called classification to designate if an email is spam or not, based on the data of an email: the sender, recipients, subject, and message body. Classification takes a set of data with known labels and learns how to label new records based on that information.
  • Clustering: Google News uses a technique called clustering to group news articles into different categories, based on title and content. Clustering algorithms discover groupings that occur in collections of data.
  • Collaborative Filtering: Amazon uses a machine learning technique called collaborative filtering (commonly referred to as recommendation), to determine which products users will like based on their history and similarity to other users.

Structure de données

Spark.MLlib est une librairie de machine learning pour Spark qui contient tous les algorithmes et utilitaires d’apprentissage classiques, comme la classification, la régression, le clustering, le filtrage collaboratif, la réduction de dimensions, en plus des primitives d’optimisation sous-jacentes. L’utilisation de tables de données résilientes (RDD), au cœur de l’efficacité́ calculatoire de Spark, rend MLlib dépendante de structures de données très contraignantes.

Vector

Création de vecteurs denses contenant les valeurs nulles :

from pyspark.mllib.linalg import Vectors # vecteur "dense"
denseVec2=Vectors.dense([1.0,0.0,2.0,4.0,0.0])

Création de vecteurs creux, seules les valeurs non nulles sont identifiées et stockées. Il faut préciser la taille du vecteur et les coordonnées de ces valeurs non nulles. C’est défini par un dictionnaire ou par une liste d’indices et de valeurs :

sparseVec1 = Vectors.sparse(5, {0: 1.0, 2: 2.0, 3: 4.0})
sparseVec2 = Vectors.sparse(5, [0, 2, 3], [1.0,2.0, 4.0])

LabeledPoint

Ce type est spécifique aux algorithmes d’apprentissage et associe un label, en fait un réel, et un vecteur. Ce label est

  • soit la valeur de la variable Y quantitative à modéliser en régression,
  • soit un code de classe : 0.0, 1.0… en classification supervisée ou discrimination.

Rating

Type de données (note d’un article par un client) spécifique aux systèmes de recommandation et donc à l’algorithme de factorisation (ALS : Alternate Least Square) disponible.

Model

La classe Model est le résultat d’un algorithme d’apprentissage qui dispose de la fonction predict() pour appliquer le modèle à une nouvelle observation ou à une table de données résiliente de nouvelles observations.

Utilisation rudimentaire de MLlib

La préparation des données est une étape essentielle à la qualité des analyses et modélisations qui en découlent. Extraction, filtrage, échantillonnage, complétion des données manquantes, correction, détection d’anomalies ou atypiques, jointures, agrégations ou cumuls, transformations (recodage, discrétisation, réduction, “normalisation”…), sélection des variables ou des features, recalages d’images ou de signaux… sont les principales procédures à mettre en œuvre de façon itérative. Par principe, la plupart de ces étapes se distribuent naturellement sur les nœuds d’un cluster en exécutant des fonctions MapReduce.

Utilisez si nécessaire la doc pour trouver la signification des paramètres transmis aux transformations et aux actions.

Kmeans

L’algorithme de classification non supervisé le plus utilisé, car le plus facile à passer à l’échelle est le Kmeans. Visualisez le fichier kmeans_data.txt dans le répertoire courant : il contient des lignes de valeurs ; on peut considérer que chaque donnée est un point de coordonnée 3D. L’algorithme des Kmeans cherche à classer les 3 premières lignes dans la première classe et les 3 dernières dans la seconde classe.

Modifiez le fichier MLlib_Kmeans.py pour pointer vers l’endroit où vous avez sauvegardé le fichier kmeans_data.txt. Si vous l’avez laissé sur le Namenode, vous pouvez utiliser file:///root/pyspark/MLlib/kmeans_data.txt. Si vous l’avez déplacer sur HDFS, alors vous pouvez utiliser input/kmeans_data.txt. Lancez alors :

spark-submit --deploy-mode client --master local[2] MLlib_Kmeans.py

Interprétez le résultat affiché sur le Terminal à la vue du code.

Régression logistique

Un exemple trivial de fouille de textes est la détection de pourriels. Le fichier spam.txt contient un spam par ligne, le fichier ham.txt un message normal par ligne. Il s’agit ensuite de prévoir le statut spam ou non d’un message.

Observez le contenu du fichier MLlib_LogReg.py :

  • Tout d’abord, il y a vectorisation des messages en utilisant la classe HashingTF qui permet de construire des vecteurs creux identifiant la fréquence de chaque mot. Un objet de cette classe est déclaré́ pouvant contenir jusqu’à̀ 10000 mots différents. Chaque message est ensuite découpé en mots et chaque mot associé à une variable. Ces actions sont traitées message par message, donc dans une étape Map.
  • Ensuite, on crée une table de données de type LabeledPoint avec 1 pour désigner un spam et 0 un message normal.
  • Ensuite, on fusionne les deux tables en une seule qui devient résiliente car mise en “cache”. C’est fort utile avant exécution d’un algorithme itératif.
  • Enfin, on estime la régression logistique puis on prédit la classe de deux messages.

Pour vérifier le fonctionnement, commencez par modifier le chemin d’accès aux fichiers spam.txt (exemples de spams) et ham.txt (exemples de courriels normaux) dans le fichier MLlib_LogReg.py. Puis, lancez la commande :

spark-submit --deploy-mode client --master local[2] MLlib_LogReg.py

Remarque : un traitement préalable supprimant les mots charnières (articles, conjonctions…) et résumant chaque mot à sa racine aurait été́ nécessaire pour une application opérationnelle.

Forêt aléatoire

Un autre système de classification qui passe l’échelle : les forêts aléatoires.

spark-submit  --deploy-mode client --master local[2] MLlib_ForetA.py

Analysez le code. Notez que les données utilisées ne sont pas extraites d’un fichier mais directement codées dans le fichier source.

Organisation des moyens de lutte contre les feux de forêt

L’objectif de ce TP est d’analyser des données pour permettre d’organiser au mieux les moyens humains et matériels pour réagir le plus rapidement en cas de départ de feu en forêt. Pour cela, on cherche à regrouper les feux de forêt ayant eu lieu les années précédentes en clusters géographiques afin de disposer de moyens de lutte aux endroits les mieux adaptés.

Les données de départ de feu proviennent des services forestiers américains et canadiens (Fire Information for Resource Management System US/Canada).

  1. Depuis le répertoire TP_BigData_ECL/TP_Spark/MLlib/, rapatriez les fichiers getshapefiles.sh, convert_shp_to_csv.py et train_Kmeans.py sur le Namenode.
  2. Appliquez la commande dos2unix getshapefiles.sh.
  3. Exécutez le script getshapefiles.sh de manière à récupérer des fichiers shapefile (GIS – Système d’information Géographique) nécessaires au TP :
./getshapefiles.sh # cela peut prendre 30 secondes

Le téléchargement correspond à 4 années de données. D’autres données sont disponibles sur le même serveur pour des années antérieures. 1. Depuis le répertoire TP_BigData_ECL/TP_Spark/MLlib/, appliquez le script convert_shp_to_csv.py sur chacun des 4 fichiers rapatriés de la manière suivante

python convert_shp_to_csv.py modis_fire_2015_365_conus.dbf modis_fire_2015_365_conus.csv
python convert_shp_to_csv.py modis_fire_2016_366_conus.dbf modis_fire_2016_366_conus.csv
python convert_shp_to_csv.py modis_fire_2017_365_conus.dbf modis_fire_2017_365_conus.csv
python convert_shp_to_csv.py modis_fire_2018_365_conus.dbf modis_fire_2018_365_conus.csv

Votre travail consiste à entraîner un classifieur Kmeans avec les données de longitude et de latitude pour

  • D’une part, obtenir le centre des clusters de départ de feu, permettant au pompier de disposer des moyens humains et matériels pour réagir au plus vite.
  • D’autre part, informer les pompiers des moyens à utiliser en cas d’un nouveau départ de feu (dont on connaît la longitude et la latitude), par prédiction du cluster le plus proche.

Bien sûr, les réponses que vous apporterez à ces questions sont nécessairement simplistes, car bien d’autres facteurs entrent en considération dans la vraie vie (notamment la morphologie du terrain) !

Avant de poursuivre, quelques remarques importantes :

  • Commencez à travailler avec un seul fichier de données, puis étendez à tous les fichiers lorsque vous aurez validé votre algorithme.
  • Attention à la conversion des chaînes de caractères en nombre : utilisez la fonction float().
  • Pour le Kmeans, utilisez les fonctions décrites dans l’API.
  • Pour vous aider, le fichier train_Kmeans.py contient la lecture d’un fichier de données et l’affichage des coordonnées des 10 premiers feux (n’oubliez pas de changer le chemin d’accès au fichier de données pour l’adapter si nécessaire). Pour le tester :
spark-submit --deploy-mode client --master local[2] train_Kmeans.py

N’oubliez pas de supprimer la première ligne qui contient les entêtes des colonnes.

  • Ne conservez que les données dont la latitude est comprise entre 42 et 50, et dont la longitude est comprise entre -124 et -110.
  • Pour l’apprentissage du classifieur, prenez un échantillon de 50% des données lues avec la méthode randomSplit(…), l’autre sera utilisé pour la prédiction.