Spark MLlib est une bibliothèque Spark d’Apache offrant des implémentations performantes de divers algorithmes d’apprentissage automatique supervisés et non supervisés. Ainsi, le framework Spark peut servir de plateforme pour développer des systèmes d’apprentissage automatique. Un modèle de Machine Learning (ML) développé avec Spark MLlib peut être associé à un pipeline de streaming crée avec Spark Structured Streaming. Cette architecture offre un faible temps de latence. Cette article intègre l’algorithme de clusternig K-means au pipeline de données développé dans les articles précédents.

Ceci est l’ultime article d’une série en quatre parties.

  • Dans la première partie, un pipeline de données est créé en Python avec Spark Structured Streaming.

  • La deuxième partie concerne la migration du pipeline vers un cluster Hadoop.

  • Dans la troisième partie, l’application PySpark a été portée et testée dans un environnement Scala Spark et des test unitaires ont été ajoutés.

  • La quatrième et dernière partie enrichie le pipeline de données avec un algorithme du regroupement par apprentissage automatique.

Pour enrichir le pipeline existant avec un algorithme de Machine Learning nous suivrons les étapes suivantes :

  • Data Engineering pour la préparation des données d’entrée

  • Développement du modèle de Machine Learning

  • Intégration du modèle de ML dans le pipeline de données en streaming

Le pipeline Spark MLlib est entraîné sur les données historiques en mode batch. Ensuite, le modèle résultant sera utilisé dans une autre application Spark Streaming. En production, les calculs des clusters sont supposés être périodiquement relancés pour mettre à jour le modèle ML.

Le code source du projet est accessible sur le dépôt GitHub adaltas/spark-mllib-streaming. Ce projet étend l’application Spark écrite en Scala du dépôt spark-streaming-scala avec un algorithme de Machine Learning k-means.

Cas d’usage pour le clustering

Le cas d’usage central à toute la série d’articles concerne les chauffeurs de Taxi de New York qui souhaitent optimiser leurs pourboires. La question à laquelle nous avons cherché à répondre dans les parties précédentes de la série était :

En tant que chauffeur de taxi, quel quartier de Manhattan choisir pour avoir un plus gros pourboire ?

Cette question impose une granularité spatiale déterminée par les quartiers de Manhattan. Un chauffeur de taxi ne peut être informé que d’un quartier où les pourboires sont importants, sans plus de précision. Suite à cette question, la solution développée précédemment reposait sur la division administrative de Manhattan.

Une granularité plus fine que les quartiers améliorerait l’utilité du service et répondrait mieux au cas d’usage. La question peut être généralisée ainsi :

Quelle région de Manhattan un chauffeur de taxi devrait-il choisir pour obtenir un pourboire élevé?

Une région peut être n’importe quelle zone de taille arbitraire. Au lieu de diviser Manhattan en quartiers, nous rechercherons de petites zones avec des pourboires élevés. Un algorithme k-means de Machine Learning sera utilisé pour obtenir des domaines d’intérêt. Chacune de ces zones est un groupe de trajets en taxi avec des pourboires élevés. Un tel groupe sera appelé un cluster par la suite.

Notez qu’un cluster peut avoir deux significations. Un cluster peut être un partitionnement de données, data clustering en anglais, par exemple un groupe de trajets en taxi. Il peut aussi être associé à un ensemble d’ordinateurs dans les systèmes distribués, comme un cluster Hadoop par exemple. Les deux termes sont pertinents au context de cet article et la signification du terme cluster devrait être clairement déductible.

Le Clustering est une tâche d’apprentissage non supervisé consistant à regrouper des objets similaires dans des catégories (les clusters). De telles tâches ne sont pas évaluables facilement, car les observations directes ne sont pas disponibles et les donneés ne sont pas annotées. Les trajets en taxi n’appartiennent pas naturellement à un groupe de solutions particulier. Le partitionnement obtenu avec le clustering peut être bon ou mauvais, selon le cas d’utilisation et l’interprétation. Le choix du nombre de cluster k, qui a le plus grand impact sur les grappes obtenues, est largement subjectif.

Obtention des données des Taxis

Le dataset utilisé est une version légèrement modifiée du dataset Taxi Data de la formation Apache Flink. Il s’agit d’un petit dataset de 100 Mo obtenu à partir de données originales des voyages en Taxi de la ville de New York.

Si vous avez un cluster Hadoop et avez suivi la deuxième partie de la série, le jeu de données brut est déjà disponible dans les répertoires /tmp/datalake/RidesRaw et /tmp/datalake/FaresRaw sur HDFS. Dans ce cas, vous pouvez utiliser ces données et ignorer le reste de cette section.

Vous pouvez suivre cette partie de la série sans cluster Hadoop. L’application Scala Spark ci-dessous crée le dataset brut nécessaire en local sur un votre ordinateur. Les données sont stockées au format de fichier Parquet et suivent le même partitionnement que dans les parties précédentes de la série.

Pour commencer, téléchargez les dataset compressés de Flink: nycTaxiRides.gz et nycTaxiFares.gz.

Ensuite, créez un fichier Scala : src/main/scala/com/adaltas/taxistreaming/DataLoader.scala dans le package com.adaltas.taxistreaming.

Compilez votre application en .jar et soumettez-le à l’exécution :

À la racine du projet, créez deux nouveaux répertoires: datalake/FaresRaw et datalake/RidesRaw. Ils contiennent les mêmes datasets comme ceux stockés sur HDFS dans les parties précédentes.

Features

Les features sont les variables d’entrées. Elles sont équivalents aux variables indépendantes en statistiques. En Machine Learning elles sont formellement définies de cette manière :

Une feature est une propriété ou une caractéristique individuelle mesurable d’un phénomène observé.

Dans notre cas, les featues sont des colonnes du dataset qui seront utilisées pour le calcul des clusters. Trois variables seront utilisées pour le regroupement: startLon, startLat et tip. Avant de pouvoir partitionner les données en fonction de ces caractéristiques, le jeu de données doit être prétraité.

Préparation des données d’entrée

Pour commencer, créez un package com.adaltas.taxistreaming.clustering contenant un fichier MainKmeans.scala. Les modèles K-means s’en serviront pour leur entraînement. Chargez les fichiers Parquet du dataset des Taxis et convertissez les en DataFrame Spark:

Ajustez les chemins de fichiers Parquet si nécessaire. Spark prend en charge les fichiers Parquet en tant que sources de données, déduit automatiquement les clés de partitionnement des fichiers Parquet et charge le schéma préservé.

Dans le fichier build.sbt, ajoutez la dépendance Spark MLlib (correspondant à votre version de Spark) :

Créez maintenant un fichier “TaxiProcessingBatch.scala” dans le package com.adaltas.taxistreaming.processing. Le code ci-dessous prend les trajets de taxi terminés dans notre région d’intérêt, Manhattan, et effectue une jointure entre les DataFrames dfRides et dfFares. Les trajets sans aucun pourboire sont filtrés. Nous créons aussi une nouvelle colonne avec l’heure de début du trajet obtenue à partir de la date de départ.

Notez qu’une nouvelle colonne tipVect est créée avec la fonction UDF vectCol (User Defined Function). Le pourboire a été inséré un vecteur MLlib de Spark. Cette conversion est nécessaire pour un traitement ultérieur avec les algorithmes de transformation de Spark MLlib.

La fonction prepareTaxiData que nous venons de définir peut être utilisée pour obtenir un DataFrame contenant un trajet en Taxi qui a débuté dans Manhattan et s’est conclu avec un pourboire. Ajoutez le à la fonction main de MainKmeans :

Mise à l’échelle des features

Dans cet article, deux types de caractéristiques sont utilisées pour le clustering : une géographique et une financière. Ces caractéristiques ont des unités incomparables et des magnitudes différentes. Par exemple, une différence de latitude de 0,05 degré est importante mais serait pratiquement négligeable par rapport au pourboire de 5 dollars. En d’autres termes, les variables géographiques seraient complètement écrasées si l’algorithme de clustering était appliqué sur le dataset en l’état.

A partir du code, nous pouvons trouver la latitude et longitude minimale de Manhattan dans les champs startLon et startLat. Chacune ne varie que de 0.1 ou 0.01 degrés.

Nous pouvons trouver le minimum et le maximum du champ tip avec la commande dfRidesManhattan.select("tip").describe().show(). Le pourboire va jusqu’à 153,35$, il varie donc bien plus que la latitude et la longitude. Cette différence poserait un problème pour l’algorithme k-means qui est sensible à la variance et les clusters seraint calculés uniquement à partir du pourboire. La mise à l’échelle permet de résoudre ce problème.

Nous ne pouvons pas simplement centrer et réduire les variables. Cela introduirait des distortions dans les coordonnées géographiques. Nous allons plutôt mettre la variable tip à la même échelle que la latitude et la longitude. Cela implique que pour avoir la véritable valeur du pourboire il faudra plus tard remettre le champ à l’échelle.

Nous pouvons utiliser le MinMaxScaler de Spark pour mettre le pourboire à l’échelle. Sa borne supérieure peut être spécifiée par le paramètre setMax. Pour utiliser le MinMaxScaler, la colonne tip doit être du type Vector. Cette conversion a déjà été faite dans la fonction prepareTaxiData où la colonne tipVect a été préparée. Le tipScaler peut être créé avec le code ci-dessous :

Les bornes des latitudes et des longitudes ont moins de 0.5 degrés d’écart, alors que les extrèmes du pourboire ont un écart de 153.35$. Pour rendre le pourboire comparable avec les caractéristiques géographiques, la valeur de setMax pour la mise à l’échelle était de 1. Cette mise à l’échelle rend intentionnelement plus important le pourboire que les informations géographiques mais aucune caractéristique ne domine désormais. Le tipScaler n’a pas encore été appliqué.

Assemblage des features

Pour utiliser l’algorithme k-means de Spark, une seule colonne de “features” de type Vector est acceptée. Actuellement, les features sont dans des colonnes différentes. Il faut les fusionner en une seule colonne de type Vector. Nous utilisons le VectorAssembler de Spark.

Ni le tipScaler ni le featureAssembler n’ont été utilisés pour le moment. Ils ont uniquement été préparés et seront incorporés dans un Pipeline MP de Spark plus tard.

K-means

L’algorithme K-means est probablement l’algorithme de classification le plus populaire. Son nom indique le nombre de k clusters à spécifier avant l’utilisation de l’algorithme. En pratique, k-means optimise le partitionnement des données pour trouver un minimum de variance intra-cluster dans tous les k clusters. Nous choisirons k = 12 comme point de départ.

Le clustering est généralement effectué sur des données statiques, mais le jeu de données taxi est une série temporelle. Ignorer l’aspect temporel de l’ensemble de données retirerait probablement tout leur sens aux clusters. La solution la plus simple consiste à diviser les trajets en segments d’une heure. Chaque segment aura un modèle de clustering distinct. Ainsi, 24 modèles k-means indépendants seront calculés, un pour chaque heure de la journée.

Vous vous demandez peut-être pourquoi le vecteur de featues ne peut pas être étendu avec une caractéristique contenant l’heure. Cela semble raisonnable, car plus la différence de temps entre les trajets est grande, plus différents il faut les considérer. Cependant, cela crée un risque d’affecter différents trajets au même cluster, uniquement parce qu’elles se sont déroulées à la même heure. Par exemple, si un groupe de trajets avait des pourboires aléatoires et une grande dispersion géographique, ils ne devraient pas être considérés comme similaires uniquement parce qu’ils se produisaient au même moment.

Un autre aspect est la mesure de similarité. Les trajets vont être jugés similaires ou différents les unes des autres. Les caractéristiques utilisées pour cette comparaison sont la longitude, la latitude et le pourboire. Cependant, diverses mesures de distance pourraient être utilisées pour comparer ces caractéristiques. Bien qu’il existe des métriques spéciales pour les séries temporelles et les données géographiques, la mesure la plus populaire est choisie : la distance euclidienne.

Spark MLlib a une implementation de KMeans, elle est initialisée comme suit:

Le pipeline Spark MLlib

La pipeline ML de Spark permet de combiner facilement plusieurs processus dans un workflow ML. Un pipeline est construit à partir des objets Transformer et Estimator. Un Transformer converts DataFrame dans un autre DataFrame. Les MinMaxScaler et VectorAssembler créés précédemment sont des exemples de Transformer. Un estimateur crée un Model, qui est un type particulier de Transformer. Dans le cas de KMeans, l’estimateur est un KMeansModel. Créez la fonction prepareKmeansPipeline dans MainKmeans à partir du package com.adaltas.taxistreaming.clustering :

Avec cette fonction, le pipeline ML souhaité est créé en une ligne de code. Testons la sur un exemple entre 10 et 11 heures du matin :

Chaque centre de cluster est un Vector Spark dont les valeurs correspondent à la longitude, latitude et le pourboire mis à l’échelle.

Evaluation des clusters

L’évaluation interne des clusters mesure la qualité des clusters. L’une des mesures pour l’évaluation interne est le score de Silhouette qui prend une valeur entre -1 et 1 dépendant de la “compacité” du cluster. Ce score est calculé ci-dessous :

Par exemple à 10h du matin, le score de Silhouette est de 0.407 pour k=12. Ce score indique une forte variance des clusters, ce qui n’est pas une bonne chose. Le score de Silhouette n’est qu’une partie de l’évaluation des clusters, un bon score ne garanti pas que les clusters soient pertinents pour notre cas d’usage. Remarquez que même des clusters de haute qualité peuvent être inutile pour le cas d’usage. Une évaluation humaine est requise. Nous allons à présent calculer la variance des clusters et visualiser nos clusters pour confirmer l’évaluation du score de Silhouette.

Calcul de la variance des clusters

La variance d’un cluster décrit la dispersion autour de la moyenne. Cela donne une idée de la taille du cluster. Créons une fonction qui prend le DataFrame predictions avec chaque course de Taxi assigne à un des k clusters dans la colonne prediction et retourne un DataFrame avec les variances intra-cluster. Ajoutons cette fonction à MainKmeans :

La fonction computeVariances utilise le Summarizer de Spark MLlib pour calculer les variances des variables startLon, startLat et de tip qui sont dans un Vecteur dans la colonne feature. Le reste du code permet de dépaquetter les Vecteurs vers des colonnes de DataFrame de haut niveau pour faciliter l’accès à la variance.

Au lieu d’utiliser les variances directement, nous allons utiliser une mesure calculée à partir de la variance: l’écart type. L’écart type est la racine carrée de la variance. L’écart type donne autant d’information que la variance mais on la préfère à cette dernière car les unités correspondent aux variables originales. Ici la variance donne des degrés au carré, l’écart type des degrés. Remarquez que les variances ont du être calculées pour obtenir l’écart type, car le Summarizer ne donne pas directement l’écart type.

Génération de GeoJSON

Pour finir l’exemple de 10h du matin, nous allons visualiser les clusters sur une carte. Nous formatterons le centre des clusters et l’écart type en GeoJSON et les visualiserons sur la carte geojson.io. Nous utiliserons circe, une librairie Scala populaire pour encoder des objets en JSON avec le format GeoJSON. Ajoutez la dépendance circe dans build.sbt :

Créez le fichier “src/main/scala/com/adaltas/taxistreaming/utils/ClustersGeoJSON.scala” dans le package com.adaltas.taxistreaming.utils. La fonction generateGeoJSON ci-dessous nous donne une sortie formatée en JSON à partir des clusters calculés précedemment.

Le centre géographique des clusters est encodé en tant que Point de geometry en format GeoJSON. L’écart type est utilisé comme rayon des cercles autour du centre des clusters. Les cercles sont plus difficiles à encoder car ils ne sont pas supportés par GeoJSON. Pour l’instant l’écart type est inclus dans les objets Properties de JSON. Plus tard, nous nous en servirons pour créer des polygones autour des clusters. La valeur tip du pourboire est aussi incluse dans les proriétés GeoJSON, elle controle l’intensité de la couleur.

La méthode generateGeoJSON produit un objet du type io.circe.Json. Utilisez cette méthode pour obtenir un fichier GeoJSON “clusters-mean.geojson” :

Dans Intellij IDEA, le projet est compilé et peut être exécuté sur Spark local pour créer le fichier “clusters-mean.geojson”. Sans IDEA, le fichier jar créé avec sbt-console échouera s’il est soumis avec spark-submit. Spark et circe dépendent tous deux de Shapeless. Malheureusement, il existe des différences entre Shapeless utilisé par Spark et Shapeless dépendant de circe. Pour résoudre ce problème, vous pouvez utiliser sbt-assembly et compiler un “gros jar” avec une dépendance masquée pour Spark ou circe. Cette tâche n’est pas simple. Si vous utilisez Intellij IDEA, vous pouvez compiler et exécuter le programme. La sortie GeoJSON n’est nécessaire que pour vérifier la validité des clusters obtenus. Il est donc acceptable d’utiliser IDEA ici et de ne pas utiliser la fonction generateGeoJSON pour la compilation du jar final.

Voici un apercu d’un fichier “clusters-mean.geojson” pour un seul cluster (les 11 autres ne sont pas montrés) :

Dans le fichier “clusters-mean.geojson”, l’écart type d’un cluster est stddev dans l’objet properties de l’objet JSON dans le tableau features (voir l’exemple ci-dessus). Nous souhaitons utiliser cette valeur pour créer un polygone de points à partir de la circonférence du cluster. Pour cela, nous utiliserons un court script Python utilisant shapely.

Le script utilise le fichier “clusters-mean.geojson” et produit un fichier “clusters-mean-stddev.geojson”. Dans le nouveau fichier GeoJSON, les rayons sont utilisés pour créer des polygones en forme de cercles autour du centre des clusters. Voici un exemple de fichier “clusters-mean-stddev.geojson” :

L’image ci-dessous montre les clusters pour l’intervalle 10h-11h. La visualisation a été obtenue en chargeant le fichier “clusters-mean-stddev.geojson” sur le site geojson.io.

L’inclinaison verticale des cercles viens du fait que nous traitons la longitude et la latitude de la même manière que le rayon. Malheureusement, l’écart type est élevé. De grandes zones autour des marqueurs indiquent que les tours à pourboire élevé commencent souvent loin du centre du cluster à laquelle ils appartiennent. Néanmoins, les résultats du regroupement semblent satisfaisants car ils abordent la question de la recherche de zones à fort pourboire.

Certains ajustements pourraient être faits pour obtenir différents clusters et de meilleurs scores de Silhouette. Plus particulièrement, le nombre de clusters pourrait être modifié. De plus, l’impact du pourboire pourrait être contrôlé avec la méthode setMax() de MinMaxScaler. Diminuer la valeur maximale du purboire donnerait plus de poids à la latitude et à la longitude, diminuant ainsi la variance géographique (au prix d’une variance supérieure du pourboire dans un cluster).

Sauvegade des modèles K-means pour chaque heure

Calculons 24 modèles k-means indépendants, un pour chaque heure de la journée. A la place de 10h, nous faisons simplement une boucle et sauvons les modèles dans un répertoire “kmeans-models” :

Remarquez que le code pour entraîner les modèles k-means n’appelle pas les fonctions computeVariances ou generateGeoJSON

Intégration du modèle ML dans le pipeline de streaming

Le pipeline de streaming développée dans les parties précédentes se décompose ainsi :

  • La donnée est envoyée vers Kafka

  • L’application Spark consomme les messages Kafka

  • Les messages Kafka sont lus et convertis en DataFrame Spark

  • Le jeu de données est sauvegardé sur HDFS

  • Un processus Spark Structures Streaming nettoie, joint, augmente et aggrège la donnée

  • Les résultats obtenus sont écrits sur HDFS et gardés en mémoire

  • Les résultats sont exposés sur JDBC pour des requêtes de faible latences

Le pipeline existant va être modifié pour prendre en compte l’ajout de l’algorithme de Machine Learning. Nous allons principalement modifier le processus de streaming de Spark Structures Streaming.

Créez un fichier “MainConsoleClustering.scala” dans le package com.adaltas.taxistreaming. L’application Spark streaming doit avoir tout les modèles k-means chargés:

Lorsqu’une course de Taxi arrive depuis le stream, l’heure de fin va servir à déterminer quel modèle de clustering utilise via la variable hourlyClusters. Cependant, hourlyClusters est un tableau de clusters (Vectors) – quel cluster est la meilleure recommendation pour un taxi libre ?

Pour déterminer le meilleur cluster pour un taxi, nous comparons les distances entre les centres des clusters et les coordonnées géographiques de la course qui s’est terminée. Définissons une fonction pour avoir la distance entre deux coordonnées :

Next, let’s put the distBetween function to use inside a UDF function picking up an attractive cluster for the Taxi driver :

Dans la fonction RecommendedLonLat, tous les clusters d’un segment horaire d’une heure sont analysés afin de faire correspondre le point d’entrée du cluster le plus proche. Pour établir qu’un cluster donné est meilleur, non seulement il doit être plus proche que le meilleur cluster actuel, mais il doit également donner un pourboire plus élevé. Enfin, les coordonnées géographiques des clusters les plus appropriés sont indiquées en tant que longitude et latitude recommandées.

Le code restant correspond au pipeline existant développé dans les parties précédentes de la série. Il copie principalement le contenu du fichier “MainConsole.scala”.

Finallement, nous pouvons intégrer un pipeline de straming avec les modèles k-means via la fonction UDF RecommendedLonLat et afficher les résultats.

Le fichier jar de l’application Scala Spark peut être compilé avec sbt-console et soumis avec spark-submit. En cas d’exécution sur le cluster Hadoop de la partie 2 de la série, soumettez-le avec :

Sinon, remplacez master par local et faites correspondre les ressources de votre machine.

Après le lancement des flux Kafka (comme expliqué dans les parties 1 et 2 de la série), les résultats finaux sont écrits sur la console:

Notez que les mises à jour périodiques des modèles k-means devraient être effectuées sur un cluster Hadoop. Cela pourrait être réalisé e.g. avec Apache Airflow en soumettant une tâche Spark avec l’application MainKmeans. Par exemple, chaque nuit, les clusters peuvent être recalculés en fonction des données de la semaine précedente.

Conclusions

La majeure partie du travail d’un projet ML concerne la préparation des données, le prétraitement, la compréhension de l’algorithme et l’évaluation. Cet article reflète cela, car beaucoup de travail a du être fait en plus d’adapter simplement l’algorithme k-means de Spark aux données du jeu de données.

Le déploiement d’un modèle ML dans un scénario temps réel est généralement difficile, mais comme le montre cet article, cela peut être simplifié. L’algorithme k-means, bien qu’entraîné sur des données statiques, a été intégré avec succès dans un pipeline de streaming Spark.