Les outils de la Data Science : Spark MLlib, mise en pratique (2/2)

Dans le précédent article, nous vous présentions les différents concepts liés à MLlib : les types de RDD à fournir, la structure des algorithmes et la démarche à adopter pour construire une chaîne de traitement de Machine Learning. Nous allons maintenant mettre en pratique ces concepts à travers plusieurs exemples basés sur les différents algorithmes qui composent MLlib.

MLlib est divisé en plusieurs packages, dont la liste exhaustive se trouve ici. Un package représente soit une catégorie d’algorithmes de Machine Learning (classification, régression, etc.), soit des outils nécéssaires à leurs usages (feature engineering, statistiques, etc.). Nous allons passer en revue dans cet article quelques-uns des principaux packages disponibles et donner des exemples d’implémentation.

La création et l’analyse exploratoire des données avec les packages random et stat

Le package random

MLlib permet la génération aléatoire de RDDs via son package random. C’est un package très utile pour faire des tests, nous permettant de créer des données selon plusieurs types de distributions: Normale, Poisson, Exponentielle, Log-Normale, Gamma et Uniforme.

Utilisons ce package pour créer un RDD[Vector], contenant un million de lignes et 4 colonnes, avec des données tirées selon une loi normale centrée réduite. On utilisera ce RDD par la suite pour présenter les différents algorithmes de Machine Learning (les résultats des algorithmes ne seront pas très bons car les données n’ont aucune structure, l’idée étant simplement de présenter comment construire une chaine de traitement en Machine Learning).

[scala]import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.random.RandomRDDs
import org.apache.spark.rdd.RDD
 
// Create a RDD[Vector] with data drawn from normal distribution
val data: RDD[Vector] = RandomRDDs.normalVectorRDD(sc, numRows = 1000000L, numCols = 4)[/scala]

Le package stat

Le package stat regroupe dans son objet Statistics des fonctions statistiques très utilisées notamment dans la partie exploratoire des données. On peut citer par exemple la fonction colStats() qui retourne une instance de MultiVariateStatisticalSummary regroupant entre autres les max, min, mean et variance pour chaque colonne d’un RDD[Vector].

[scala]import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.stat.{MultivariateStatisticalSummary, Statistics}

// Compute column summary statistics.
val summary: MultivariateStatisticalSummary = Statistics.colStats(data)
println(summary.mean) // a dense vector containing the mean value for each column
println(summary.variance) // column-wise variance
println(summary.max) // highest number for each column[/scala]

Appliquées au RDD généré dans l’étape précédente, on devrait retrouver des valeurs proches de 0 pour la moyenne et de 1 pour la variance.

On trouve de plus une fonction de calcul de la corrélation entre deux RDD. Les corrélations supportées sont celles de Pearson et de Spearman. Si on ne spécifie qu’un seul RDD, la fonction calcule l’autocorrélation.

[scala]import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.stat.Statistics

// Calculate the correlation matrix using Pearson’s method. Use "spearman" for Spearman’s method.
// If a method is not specified, Pearson’s method will be used by default.
val correlMatrix: Matrix = Statistics.corr(data, "pearson")[/scala]

Pour notre RDD, la matrice retournée doit donner des 1 sur la diagonale, et des valeurs proches de 0 sur le reste (pas de corrélation entre les différentes colonnes).

Pour compléter ce panel, on trouve aussi des fonctions relatives aux tests d’hypothèses, comme le test du chi-squared (chiSqTest).

Les algorithmes d’apprentissage supervisé avec les packages classification et regression

Comme leurs noms l’indiquent, les packages classification et regression regroupent tous les principaux algorithmes liés à l’apprentissage supervisé (voir le précédent article sur MLlib pour un rappel sur le Machine Learning) et vont donc demander l’utilisation de RDD[LabeledPoint]. Les deux types ont pour objectif de prédire une variable grâce à des features en utilisant un training set contenant les vrais labels. La différence entre les deux est qu’en classification, on cherche à prédire une variable discrète (ex: spam / non spam), alors qu’elle est continue en régression. Voyons maintenant comment sont utilisés quelques algorithmes classiques de Machine Learning présents dans ces packages.

Un exemple en régression: LinearRegressionWithSGD

La régression linéaire est l’algorithme le plus classique pour des problématiques de régression. En MLlib, il est implémenté en utilisant de la Descente de Gradient Stochastique, d’où son nom. Son objectif est de minimiser la fonction de coût suivante:

Avec:

• w le vecteur des paramètres appris par l’algorithme (correspondant globalement au poids de chaque caractéristique)
• x un vecteur de caractéristiques (un point dans l’espace)
• y la valeur réelle observée

Il est possible d’y ajouter des termes de régularisation, donnant lieu aux algorithmes dits de Ridge Regression ou de Lasso.

Voici un exemple d’implémentation, utilisant comme données d’entrée le RDD créé précédemment. Un autre RDD nommé label est aussi créé, il va représenter notre variable cible. Il est créé selon une loi uniforme (double entre 0 et 1).

[scala]import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.random.RandomRDDs
import org.apache.spark.mllib.regression.{LinearRegressionWithSGD, LinearRegressionModel, LabeledPoint}
import org.apache.spark.mllib.evaluation.RegressionMetrics
import org.apache.spark.rdd.RDD

// Create a RDD[Vector] with data drawn from normal distribution
val data: RDD[Vector] = RandomRDDs.normalVectorRDD(sc, numRows = 1000000L, numCols = 4)

// Create a RDD of labels for regression
val label: RDD[Double] = RandomRDDs.uniformRDD(sc, 1000000L)

// Create a RDD[LabeledPoint]
val dataWithLabel: RDD[LabeledPoint] = data.zip(label).map(point => LabeledPoint(point._2, point._1))
dataWithLabel.cache()

// Building the model
val numIterations = 100
val model: LinearRegressionModel = LinearRegressionWithSGD.train(dataWithLabel, numIterations)

// Evaluate model on training examples
val predictionAndLabels: RDD[(Double, Double)] = dataWithLabel.map { case LabeledPoint(label, features) =>
val prediction: Double = model.predict(features)
(prediction, label)
}

// Compute Mean Square Error
val metrics: RegressionMetrics = new RegressionMetrics(predictionAndLabels)
val MSE = metrics.meanSquaredError
println("training Mean Squared Error = " + MSE)[/scala]

La démarche est classique, et on la retrouvera quasiment systématiquement:

• Création du RDD sous le bon format
• Construction et entraînement du modèle
• Prédiction et évaluation des prédictions faites

On remarquera que l’on a préféré la méthode train() de LinearRegressionWithSGD en mettant le nombre d’itérations en paramètre de la méthode, plutôt que d’utiliser la méthode run() avec son setter associé. De plus, il est fortement recommandé d’utiliser la méthode cache() pour monter notre RDD en mémoire. De nombreux algorithmes de Machine Learning nécessitent plusieurs itérations sur les données, leur montée en mémoire va donc réduire fortement le temps de calcul.

Pour les algorithmes de régression, il est possible d’évaluer les modèles à l’aide de la classe RegressionMetrics, qui permet d’obtenir notamment la mean squared error.

Un exemple en classification: LogisticRegressionWithLBFGS

Contrairement à ce que son nom peut sembler l’indiquer, la régression logistique est un algorithme de classification utilisé fréquemment pour prédire une réponse (binaire ou multi-classes). La fonction de coût qu’il veut minimiser est la suivante :

Son implémentation en MLlib se fait soit par Descente de Gradient Stochastique (LogisticRegressionWithSGD) ou par l’algorithme Limited-Memory Broyden-Fletcher-Goldfarb-Shanno (LogisticRegressionWithLBFGS), qui est généralement préféré. Cette dernière implémentation ne possède pas de méthode train(), nous aurons donc recours à la méthode run() et à l’appel des setters associés à la classe.

[scala]import org.apache.spark.mllib.classification.{LogisticRegressionModel, LogisticRegressionWithLBFGS}
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.random.RandomRDDs
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.rdd.RDD

// Create a RDD[Vector] with data drawn from normal distribution
val data: RDD[Vector] = RandomRDDs.normalVectorRDD(sc, numRows = 1000000L, numCols = 4)

// Create a RDD of labels for classification (5 classes)
val label: RDD[Double] = RandomRDDs.uniformRDD(sc, 1000000L).map(x => (x*5).floor)

// Create a RDD[LabeledPoint]
val dataWithLabel: RDD[LabeledPoint] = data.zip(label).map(point => LabeledPoint(point._2, point._1))

// Split data into training (75%) and test (25%).
val splits: Array[RDD[LabeledPoint]] = dataWithLabel.randomSplit(Array(0.75, 0.25), seed = 11L)
val training: RDD[LabeledPoint] = splits(0).cache()
val test: RDD[LabeledPoint] = splits(1)

// Run training algorithm to build the model
val model: LogisticRegressionModel = new LogisticRegressionWithLBFGS().setNumClasses(10).run(training)

// Compute raw scores on the test set
val predictionAndLabels: RDD[(Double, Double)] = test.map { case LabeledPoint(label, features) =>
val prediction: Double = model.predict(features)
(prediction, label)
}

// Get evaluation metrics
val metrics: MulticlassMetrics = new MulticlassMetrics(predictionAndLabels)
val precision: Double = metrics.precision
println("Precision = " + precision)[/scala]

La démarche appliquée est la même que dans l’exemple précédent, avec en plus une étape de splitting du RDD, ce qui est recommandé en apprentissage supervisé. L’apprentissage se fait alors sur le training set, et les prédictions et évaluations se font sur le test set. Cela permet d’avoir une meilleure idée de la capacité de généralisation de notre algorithme à de nouvelles données sur lesquelles il ne s’est pas entraîné.

Pour les algorithmes de classification, il est possible d’évaluer les modèles à l’aide de la classe MulticlassMetrics (dans le cas où il y a plus de deux classes), qui permet d’obtenir notamment la précision ou la matrice de confusion. Pour une classification binaire, il faut plutôt utiliser BinaryClassificationMetrics, qui a un fonctionnement similaire mais possède des métriques supplémentaires spécifiques à ce type de classification.

Autres algorithmes

Voici quelques autres algorithmes présents dans les packages classification et regression:

•  Classification
  • NaiveBayes: Algorithme qui suppose l’indépendance de chaque caractéristique du jeu de données pour faciliter le calcul de probabilité d’appartenance à une classe. Il est par exemple utilisé dans la classification de spams.
  • SVMWithSGD: C’est une généralisation des classifieurs linéaires plus classiques. Il donne souvent de bons résultats en pratique et est particulièrement utilisé lorsqu’il faut travailler avec des données à grandes dimensions. Il a de plus l’avantage de ne nécessiter que d’un faible nombre d’hyper-paramètres (les paramètres qu’il faut tuner pour améliorer les résultats de l’algorithme).
• Régression
  • LassoWithSGD: Algorithme de régression linéaire pour lequel on ajoute une pénalisation du premier ordre (on ajoute la somme des valeurs absolues des paramètres à la fonction de coût) pour faire de la régularisation et rendre l’algorithme plus robuste et moins enclin à de l’overfitting (apprentissage par coeur du training set)
  • RidgeRegressionWithSGD: Identique au précédent, mais la pénalisation est cette fois du second ordre (on ajoute la somme du carré des paramètres à la fonction de coût).

Leur utilisation se fait exactement de la même manière que pour les deux algorithmes précédemment présentés. Il suffit de regarder dans la documentation les différents paramètres qui leurs sont propres.

Le package tree pour les algorithmes fonctionnant à la fois en régression et en classification

Les algorithmes basés sur des arbres de décision (DecisionTree, GradientBoostedTree, RandomForest) pouvant être utilisés à la fois en classification et en régression, ils bénéficient d’un package tree qui leur est propre.  Ils permettent de plus d’utiliser des variables catégorielles (il faut alors spécifier à l’algorithme quelles sont les variables catégorielles en question).

DecisionTree est un algorithme très commun en Machine Learning car il est très interprétable et n’oblige pas à faire un scaling préalable des données. Concrètement, cet algorithme fonctionne en pratiquant de manière récursive des partitions binaires dans l’espace des caractéristiques. A chaque étape de splitting, l’algorithme va choisir la caractéristique qui va maximiser le gain en information après le split, c’est-à-dire minimiser notre incertitude sur la variable à prédire dans les subsets créés.

RandomForest et GradientBoostedTree sont des méthodes dites d’ensemble. Le principe n’est plus de construire un unique arbre de décision, mais un ensemble, soit en parallèle (RandomForest), soit les uns après les autres sur les résidus des précédents (GradientBoostedTree). Ce sont des algorithmes très puissants, donnant souvent de très bons résultats.

En plus de la méthode classique run(), il est possible pour les algorithmes DecisionTree et RandomForest d’utiliser les méthodes trainClassifier() ou trainRegressor(), qui ont des paramètres différents. Voici un exemple de classification avec un DecisionTree.