Jusqu'à présent, on a vu avec Spark des outils qui permettent de réaliser facilement des opérations bas niveau de manière distribuée. Ces briques élémentaires nous permettent de réécrire des algorithmes entiers pour les distribuer sur des clusters de calcul. Cependant il peut être fastidieux de réécrire des algorithmes complexes pour lesquels il existe déjà des implémentations populaires mais non distribuées. MLlib est une librairie de machine learning pour Spark qui résout ce problème. De la même manière, on pourrait réécrire des méthodes de traitement de données structurées mais les concepteurs de Spark ont pensé à tout et ont créé pour ça Spark SQL.
Spark SQL
Commençons par parler de Spark SQL, qui va nous permettre d'introduire un nouveau type d'objet : les DataFrames. Il s'agit d'un type très similaire aux RDD, à la différence qu'il permet de stocker de manière distribuée des données structurées, là où les RDD nous permettent de stocker des données non structurées. Pas de panique, on va voir qu'en pratique les différences sont mineures.
Quel est l'intérêt de stocker des données structurées de manière distribuées ? Cela va nous permettre d'analyser nos données à l'aide de requêtes similaires à des requêtes SQL — souvenez-vous que "SQL" signifie Structured Query Language. Structurer des données massives va donc vous permettre de mettre à disposition de vos utilisateurs des données qu'ils pourront exploiter comme s'ils effectuaient des requêtes sur une base de données relationnelle classique, et ils n'auront pour ça pas à apprendre un nouveau langage de requête.
Lecture de données
Pour comprendre Spark SQL et les DataFrames, le mieux est encore de travailler sur un exemple. Nous allons analyser une base de données contenant un échantillon de la population mondiale.
Commençons par télécharger nos données :
$ wget https://s3-eu-west-1.amazonaws.com/course.oc-static.com/courses/4297166/agents.json
$ head -5 agents.json
{"id":227417393,"longitude":100.85840672174572,"latitude":33.15219798270325,"country_name":"China","sex":"Male"}
{"id":6821129477,"longitude":-72.43795260265814,"latitude":19.325567983697297,"country_name":"Haiti","sex":"Female"}
{"id":2078667700,"longitude":80.85636526088884,"latitude":23.645271492037235,"country_name":"India","sex":"Female"}
{"id":477556555,"longitude":93.33604038078953,"latitude":33.45864668881662,"country_name":"China","sex":"Female"}
{"id":1379059984,"longitude":80.7728698035823,"latitude":28.816938290678692,"country_name":"India","sex":"Female"}Nous allons faire des expériences avec Spark SQL dans un Spark Shell :
$ ./code/spark-2.3.1-bin-hadoop2.7/bin/pyspark
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/__ / .__/\_,_/_/ /_/\_\ version 2.3.1
/_/
Using Python version 3.6.5 (default, Apr 1 2018 05:46:30)
SparkSession available as 'spark'.Peut-être aviez-vous déjà remarqué ce message lors du lancement d'un Spark Shell :SparkSession available as 'spark'? Effectivement, un objetsparkde typepyspark.sql.session.SparkSessionest disponible dans le shell :
>>> spark
<pyspark.sql.session.SparkSession object at 0x7fdbb735ea50>Cet objet nous permet d'instancier des objets de typeDataFrame, de la même manière que leSparkContextnous permettait d'instancier desRDD:
>>> agents = spark.read.json("agents.json")
>>> agents
DataFrame[country_name: string, id: bigint, latitude: double, longitude: double, sex: string]On remarque que le schéma des données a été automatiquement inféré, et nous disposons donc maintenant d'une interface structurée vers le contenu de notre fichier. Mais attention ! Les données ne sont pas chargées en mémoire pour autant. Comme pour les RDD, les données ne seront lues que lorsque ça sera nécessaire, en suivant le même principe d'évaluation paresseuse.
Écrire des requêtes
Continuons notre exemple précédent pour analyser le contenu de notre base d'agents :
>>> french_agents = agents.filter(agents.country_name == "France")
>>> french_agents
DataFrame[country_name: string, id: bigint, latitude: double, longitude: double, sex: string]L'opérationfilterrenvoie un objet de typeDataFrame, tout comme les transformations sur les objets de typeRDDrenvoient un objet de typeRDD. De fait, l'opérationfilterne fait pas grand chose, en l'état, puisqu'elle n'a pas encore été évaluée. Pour l'évaluer, il faut réaliser une action :
>>> french_agents.count()
94La récupération du premier objet duDataFrameconstitue aussi une action et déclenche donc l'évaluation defilter:
>>> agent = french_agents.first()
>>> agent
Row(country_name=u'France', id=5130782577, latitude=-0.21142875508479517, longitude=-0.003950214433749498, sex=u'Female')On peut voir que les objets retournés par leDataFramesont de typeRow. On accède aux valeurs des colonnes de cet objet comme à des attributs standards :
>>> print(agent.country_name, agent.id)
France 5130782577En réalité, l'API des DataFrames est conçue comme un ORM ("Object-Relational Mapping"), et nous pouvons réaliser des requêtes SQL en enchaînant des appels de méthodes :
>>> agents.filter(agents.country_name == "France").filter(agents.latitude < 0).count()
41La même opération peut être réalisée à l'aide d'opérateurs booléens :
>>> agents.filter((agents.country_name == "France") & (agents.latitude < 0)).count()
41Le contenu d'un DataFrame peut être visualisé à l'aide de la méthodeshow(), qui est aussi une action :
>>> agents.limit(5).show()
+------------+----------+------------------+------------------+------+
|country_name| id| latitude| longitude| sex|
+------------+----------+------------------+------------------+------+
| China| 227417393| 33.15219798270325|100.85840672174572| Male|
| Haiti|6821129477|19.325567983697297|-72.43795260265814|Female|
| India|2078667700|23.645271492037235| 80.85636526088884|Female|
| China| 477556555| 33.45864668881662| 93.33604038078953|Female|
| India|1379059984|28.816938290678692| 80.7728698035823|Female|
+------------+----------+------------------+------------------+------+On peut également réaliser des requêtes SQL sans passer par l'ORM en créant une vue temporaire associée à un DataFrame :
>>> agents.createTempView("agents_table")
>>> spark.sql("SELECT * FROM agents_table ORDER BY id DESC LIMIT 10").show()
+-----------------+----------+-------------------+-------------------+------+
| country_name| id| latitude| longitude| sex|
+-----------------+----------+-------------------+-------------------+------+
| French Polynesia|7170821229|-15.004219445056265|-140.01650828107668| Male|
| Cabo Verde|7167692449| 16.00676587564149| -23.90898775675409| Male|
| Suriname|7166451460| 4.008871704322331| -55.97275746253122|Female|
| Suriname|7166235088| 3.96442417744574|-56.077562332679605|Female|
| Macau|7166034642| 21.944944804684596| 114.02447154998114|Female|
| Montenegro|7164357515| 42.32131745506727| 19.168822000529843| Male|
|Equatorial Guinea|7163867872| 3.651402073464487| 9.913739020397387|Female|
| Bhutan|7163256789| 27.419739555133912| 90.29001406759927|Female|
| Bhutan|7163004645| 27.281480489455422| 90.17405662751794| Male|
| Bhutan|7162877973| 27.37149433886258| 90.38882928596311| Male|
+-----------------+----------+-------------------+-------------------+------+La vueagents_tableainsi créée est temporaire ; vous pouvez la remplacer à l'aide de la méthodecreateOrReplaceTempView. Pour plus de précisions concernant la gestion de vues temporaires ou globales, n'hésitez pas à consulter la documentation officielle.
Il faut bien comprendre que tous les principes essentiels qu'on a vu sur les RDD dans le chapitre précédent s'appliquent de la même manière aux DataFrames. Par exemple, on peut stocker en cache un DataFrame à l'aide de la méthodepersist:
>>> agents.persist()
DataFrame[country_name: string, id: bigint, latitude: double, longitude: double, sex: string]Et les RDD ne sont jamais loin puisque le RDD associé à un DataFrame est accessible via la propriétérdd:
>>> agents.rdd.filter(lambda row: row.country_name == "France").count()
94Il est donc très aisé de convertir un DataFrame en RDD. Par ailleurs, unSparkSessionpeut convertir un RDD contenant des objets de typeRowen DataFrame :
>>> from pyspark.sql import Row
>>> rdd = sc.parallelize([Row(name="Alice"), Row(name="Bob")])
>>> spark.createDataFrame(rdd)
DataFrame[name: string]Pourquoi ce long laïus préambulatoire sur les DataFrames ? Et bien tout d'abord, les DataFrames permettent de réaliser des requêtes sur des données structurées. Ceci vous permet de mettre à disposition de vos équipes de business intelligence ou de data scientists une interface dotée d'un langage de requêtes dont ils sont déjà familiers. Ensuite, les DataFrames sont fortement utilisées dans Spark pour faire du machine learning, comme on va le voir maintenant.
Spark ML
Savez-vous ce qu'est le machine learning ? On est censé parler d'"apprentissage automatique" en français mais cette appellation n'a jamais été très populaire... Le machine learning est la science que pratiquent les data scientists : d'un point de vue très formel, le machine learning permet de réaliser des prédictions sur des données à partir d'autres données, ce qui constitue, vous en conviendrez, une définition assez aride. Pour être plus concret, voyons quelques exemples de problèmes de machine learning :
Estimer la fréquentation d'un lieu touristique le mois prochain en fonction de sa fréquentation des dix dernières années.
Décrire le contenu d'une image à partir des valeurs des pixels de l'image et du contenu d'un million d'autres images.
Détecter les groupes d'utilisateurs les plus soudés d'un réseau social.
Ces exemples sont des problèmes typiques auxquels sont confrontés les data scientists. Pour en savoir plus, je vous conseille de consulter l'excellent cours Initiez-vous au machine learning. Prenez votre temps je ne bouge pas d'ici :)
Vous n'avez pas forcément besoin de devenir un data scientist aguerri pour réaliser des calculs distribués avec Spark. Cependant, si vous êtes amenés à travailler en tant que data architect (ce qu'on vous souhaite !) vous aurez à collaborer avec des data scientists et à comprendre les enjeux du machine learning. En particulier, les deux problèmes les plus fréquents rencontrés par les data scientists concernent :
La rapidité d'exécution de leurs calculs.
Le volume de données qu'ils peuvent utiliser pour réaliser l'apprentissage de leurs algorithmes.
En effet, la réalisation de prédictions sur des données nécessite (presque toujours) de calculer les paramètres d'un modèle. Par exemple, si l'on cherche à créer un modèle de prédiction de la météo, il faut mesurer l'impact quantitatif d'un vent du nord sur l'hygrométrie. Pour distinguer des images de chatons d'images de chiots, il faut déterminer quels sont les attributs visuels qui différencient les deux espèces. Et pour apprendre ces modèles il faut des données, beaucoup de données. Plus il y a de paramètres dans le modèle, plus il faut de données pour réaliser son apprentissage. Et les data scientists ont tendance à créer des modèles complexes nécessitant beaucoup de paramètres. Ils ne font pas ça exprès pour vous embêter, c'est juste que la réalité c'est complexe et difficile à modéliser.
Par ailleurs, même si un modèle nécessite relativement peu de paramètres, il sera d'autant plus réaliste que beaucoup de données auront été utilisées pour le créer. Par exemple, pour ajuster précisément les paramètres de votre modèle météorologique, vous avez besoin d'un historique le plus ancien possible des relevés de température et d'hygrométrie.
Bref, l'augmentation du nombre de paramètres et la nécessité d'ajuster avec précision les paramètres d'un modèles sont deux phénomènes qui concourent à utiliser le plus de données possible pour réaliser l'apprentissage d'un modèle. Le problème, pour les data scientists, c'est que ces besoins en données excèdent rapidement les ressources en calcul et en mémoire d'une unique machine. Et votre rôle, en tant que data architect, est de permettre malgré tout la réalisation de ces calculs complexes... Vous l'aurez deviné, en mettant à disposition une infrastructure de calcul distribuée !
Mais trève de bavardages, rentrons dans le vif du sujet avec l'apprentissage d'un modèle de classification.
Classification automatique de textes
Nous allons créer un modèle de classification de textes capable de prédire quel est le sujet d'une conversation. Enfin, pas n'importe quelle conversation, restons simples. Nous allons utiliser un jeu de données public contenant des conversations ayant eu lieu dans des newsgroups Usenet (à chanter sur l'air de "Je vous parle d'un temps que les moins de vingt ans ne peuvent pas connaaaaaaaaîtreuuuuux").
La documentation relative au jeu de données est donnée ici. Chacun des 18821 messages appartient à un newsgroup, parmi 20 possibles. Chaque newsgroup parle d'un sujet différent : il y a des sujets légers comme les voitures, le baseball, ou le hockey, ou beaucoup plus sérieux, comme le Moyen Orient, la religion ou Windows.
Vous êtes prêts ? C'est parti. Commençons, par télécharger les données d'apprentissage et de test :
$ wget http://ana.cachopo.org/datasets-for-single-label-text-categorization/20ng-train-all-terms.txt
$ wget http://ana.cachopo.org/datasets-for-single-label-text-categorization/20ng-test-all-terms.txtLa différence entre les données d'apprentissage et de test, c'est que les premières peuvent être utilisées pour l'apprentissage du modèle, tandis que les secondes ne peuvent servir que pour son évaluation. S'il s'agissait d'un modèle météo, on aurait dans les données d'apprentissage toutes les données météo du passé, tandis que les données de test concerneraient la journée en cours.
Nous allons charger ces données sous forme de RDD et les convertir en DataFrames, qui est le format recommandé lors de l'utilisation de Spark ML :
from pyspark.sql import Row
def load_dataframe(path):
rdd = sc.textFile(path)\
.map(lambda line: line.split())\
.map(lambda words: Row(label=words[0], words=words[1:]))
return spark.createDataFrame(rdd)
train_data = load_dataframe("20ng-train-all-terms.txt")
test_data = load_dataframe("20ng-test-all-terms.txt")Dans le jeu de données téléchargé, chaque message est un ensemble de mots séparés par des espaces sur une seule ligne. La ponctuation a déjà été supprimée et tous les mots sont en minuscule. Le premier mot indique le newsgroup auquel appartient le message. On dit que c'est le label du message (le but étant, rappelons-le, de prédire les labels des messages de test).
Nous allons commencer par créer une représentation de nos messages sous la forme d'un sac de mots (bag of words). Un sac de mots est une représentation d'un texte sous la forme d'un histogramme indiquant le nombre d'occurences de chaque mot. On appelle ça un "sac" parce que l'ordre des mots est perdu dans cette représentation. C'est en fait la même chose qu'un nuage de mots. A titre d'exemple, voilà le nuage de mots de l'Iliade :
Réalisé avec WordItOut
Dans un sac de mots, un texte est représenté par un tableau d'entiers de longueur égale à celle du dictionnaire, donc potentiellement de très grande taille. Par exemple, admettons que la langue anglaise contienne 100000 mots. Si je cherche à représenter sous la forme d'un sac de mots anglais la phrase "hello world" il me faudra stocker un vecteur de taille 100000 ne contenant que des zéros, sauf à deux emplacements correspondant à "hello" et "world", où on stockera des 1.
Avec Spark ML, cette opération se fait à l'aide d'un CountVectorizer :
from pyspark.ml.feature import CountVectorizer
vectorizer = CountVectorizer(inputCol="words", outputCol="bag_of_words")Cet objetvectorizerest unEstimator: lorsque l'on appelle sa méthodefitsur unDataFrameil produit unTransformer:
vectorizer_transformer = vectorizer.fit(train_data)Le rôle d'unTransformerest... de transformer les données :
train_bag_of_words = vectorizer_transformer.transform(train_data)
test_bag_of_words = vectorizer_transformer.transform(test_data)Dans le cas présent,vectorizer_transformertransforme unDataFramecontenant une colonnewordsen un autreDataFramecontenant une colonnebag_of_words. Comme vous l'avez peut-être deviné, lorsqu'il est appliqué à unDataFramela méthodetransformest évaluée de manière paresseuse, comme pour les transformations réalisées sur les RDD.
Nous avons maintenant dans la colonnebag_of_wordsdes DataFrames d'apprentissage et de test une représentation sous forme de vecteur d'entiers de nos textes. Il nous reste à apprendre un classifieur. Nous allons utiliser le classifieur NaiveBayes qui a l'avantage d'être multi-classes : ça signifie que ce modèle peut être utilisé pour différencier plus de deux classes différentes, comme c'est le cas ici. Pour d'autres modèles, il n'est possible de réaliser leur apprentissage que sur des jeux de données contenant deux classes.
Cependant, pour réaliser l'apprentissage du classifieurNaiveBayes, nous avons besoin de labels sous la forme de nombres flottants, et non de chaînes de caractères comme c'est le cas pour l'instant. En effet, la colonnelabelcontient pour l'instant les noms des newsgroups auxquels appartiennent les messages :
>>> train_data.select("label").distinct().sort("label").show(truncate=False)
+------------------------+
|label |
+------------------------+
|alt.atheism |
|comp.graphics |
|comp.os.ms-windows.misc |
|comp.sys.ibm.pc.hardware|
|comp.sys.mac.hardware |
|comp.windows.x |
|misc.forsale |
|rec.autos |
|rec.motorcycles |
|rec.sport.baseball |
|rec.sport.hockey |
|sci.crypt |
|sci.electronics |
|sci.med |
|sci.space |
|soc.religion.christian |
|talk.politics.guns |
|talk.politics.mideast |
|talk.politics.misc |
|talk.religion.misc |
+------------------------+On pourrait réaliser une associationlabel --> label_indexà la main en récupérant la liste des labels, mais Spark ML fournit unEstimatorqui fait précisément cela :
from pyspark.ml.feature import StringIndexer
label_indexer = StringIndexer(inputCol="label", outputCol="label_index")
label_indexer_transformer = label_indexer.fit(train_bag_of_words)
train_bag_of_words = label_indexer_transformer.transform(train_bag_of_words)
test_bag_of_words = label_indexer_transformer.transform(test_bag_of_words)Nous pouvons maintenant réaliser l'apprentissage de notre modèle de classification :
from pyspark.ml.classification import NaiveBayes
classifier = NaiveBayes(
labelCol="label_index", featuresCol="bag_of_words", predictionCol="label_index_predicted"
)
classifier_transformer = classifier.fit(train_bag_of_words)Appliquons ce modèle à nos données de test :
test_predicted = classifier_transformer.transform(test_bag_of_words)Examinons le résultat :
>>> test_predicted.select("label_index", "label_index_predicted").limit(10).show()
+-----------+---------------------+
|label_index|label_index_predicted|
+-----------+---------------------+
| 17.0| 17.0|
| 17.0| 17.0|
| 17.0| 17.0|
| 17.0| 17.0|
| 17.0| 17.0|
| 17.0| 17.0|
| 17.0| 19.0|
| 17.0| 17.0|
| 17.0| 17.0|
| 17.0| 17.0|
+-----------+---------------------+C'est pas si mal ! Sur les dix premiers messages notre classifieur ne s'est trompé qu'une fois.
La précision de notre modèle peut être évaluée facilement à l'aide de la classe MulticlassClassificationEvaluator :
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluator = MulticlassClassificationEvaluator(labelCol="label_index", predictionCol="label_index_predicted", metricName="accuracy")
accuracy = evaluator.evaluate(test_predicted)
print("Accuracy = {:.2f}".format(accuracy))On obtient une précision de 0.80, ce qui signifie que 80% de nos données sont correctement classifiées. Pas mal ! D'après la page à partir de laquelle nous avons téléchargé les données, la meilleure précision obtenue est de 0.8284.
Récapitulons le code que nous avons rédigé sous la forme d'une application Spark :
from pyspark import SparkContext
from pyspark.sql import SparkSession
sc = SparkContext()
spark = SparkSession.builder.getOrCreate()
from pyspark.ml.classification import NaiveBayes
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import CountVectorizer, StringIndexer
from pyspark.sql import Row
def load_dataframe(path):
rdd = sc.textFile(path)\
.map(lambda line: line.split())\
.map(lambda words: Row(label=words[0], words=words[1:]))
return spark.createDataFrame(rdd)
# Load train and test data
train_data = load_dataframe("20ng-train-all-terms.txt")
test_data = load_dataframe("20ng-test-all-terms.txt")
# Learn the vocabulary of our training data
vectorizer = CountVectorizer(inputCol="words", outputCol="bag_of_words")
vectorizer_transformer = vectorizer.fit(train_data)
# Create bags of words for train and test data
train_bag_of_words = vectorizer_transformer.transform(train_data)
test_bag_of_words = vectorizer_transformer.transform(test_data)
# Convert string labels to floats
label_indexer = StringIndexer(inputCol="label", outputCol="label_index")
label_indexer_transformer = label_indexer.fit(train_bag_of_words)
train_bag_of_words = label_indexer_transformer.transform(train_bag_of_words)
test_bag_of_words = label_indexer_transformer.transform(test_bag_of_words)
# Learn multiclass classifier on training data
classifier = NaiveBayes(
labelCol="label_index", featuresCol="bag_of_words", predictionCol="label_index_predicted"
)
classifier_transformer = classifier.fit(train_bag_of_words)
# Predict labels on test data
test_predicted = classifier_transformer.transform(test_bag_of_words)
# Classifier evaluation
evaluator = MulticlassClassificationEvaluator(
labelCol="label_index", predictionCol="label_index_predicted", metricName="accuracy"
)
accuracy = evaluator.evaluate(test_predicted)
print("Accuracy = {:.2f}".format(accuracy))Comme vous l'avez peut-être remarqué, cette application est très verbeuse. A quatre reprises nous créons unEstimatorsur lequel nous appelons la méthodefitqui retourne unTransformer, sur lequel nous appliquons à son tour la méthodetransform. Cette succession d'opérations est classique en machine learning. Pour simplifier le code des applications, Spark ML propose de formaliser cet enchaînement sous la forme d'unPipeline.
Simplifiez-vous la vie avec les Pipelines
Plutôt que d'instancier chaqueEstimator, d'appeler la méthodefitpuistransformséparément, nous allons instancier un Pipeline dont le rôle va être d'appliquer une série d'Estimator à nos données. UnPipelines'instancie de la manière suivante :
from pyspark.ml import Pipeline
pipeline = Pipeline([estimator1, estimator2, transformerr1, estimator3, ...])Lorsqu'on appelle la méthodefitsur unPipeline, les Estimators sont changés en Transformers, et les Transformers ne changent pas.
Bref, sans plus tarder, voyons comment unPipelinepermet de simplifier notre code :
vectorizer = CountVectorizer(inputCol="words", outputCol="bag_of_words")
label_indexer = StringIndexer(inputCol="label", outputCol="label_index")
classifier = NaiveBayes(
labelCol="label_index", featuresCol="bag_of_words", predictionCol="label_index_predicted",
)
pipeline = Pipeline(stages=[vectorizer, label_indexer, classifier])
pipeline_model = pipeline.fit(train_data)
test_predicted = pipeline_model.transform(test_data)Plus simple, n'est-ce pas ? L'intérêt de cette formalisation est qu'elle nous permet d'ajouter facilement des étapes au Pipeline. L'application réécrite à l'aide de pipelines peut être consultée (et améliorée !) sur le dépôt de code github consacré au cours. A titre d'exercice, ajoutez une pondération TF-IDF à notre représentation de sac de mots. Il se pourrait bien que vous battiez le record de précision pour notre jeu de données...
Un dernier mot sur les RDD...
À l'heure où nous parlons, l'API de Spark ML est en cours de réécriture. L'API à base de DataFrames que nous avons vue est la version moderne, tandis qu'il existe une version plus ancienne qui opère, elle, sur des RDD. Comme indiqué dans la documentation de Spark l'API à base de RDD est vouée à disparaître dans Spark 3.0, et aucune nouvelle fonctionnalité n'y sera ajoutée d'ici là. Cependant il existe encore des fonctionnalités qui ne sont pas présentes dans l'API à base de DataFrames. Par exemple, la réduction de dimension par PCA (Principal Component Analysis) et la classification par SVM (Support Vector Machine), pour ne citer qu'eux, opèrent encore sur des RDD.
Les fonctionnalités à base de RDD sont disponibles dans le modulepyspark.mllibet non danspyspark.mlcomme on l'a vu jusqu'à présent. Ceci peut vous aider à savoir sur quel type de données opèrent les fonctions que vous utilisez.
Nous n'allons pas couvrir l'API de Spark MLlib dans ce cours, même si nous vous encourageons à consulter sa documentation. Toutes les notions que nous avons vues ensemble dans les chapitres précédents sur les RDD de ce cours restent valables dans Spark MLlib. Vous devez seulement savoir que les classifieurs de Spark MLlib opèrent sur des RDD contenant des objets de type LabeledPoint.
UnLabeledPointest un objet qui contient un label, sous la forme d'un flottant, et un vecteur d'attributs. Pour reprendre l'exemple précédent sur la classification de textes, voici comment convertir un DataFrame contenant unlabel_indexet unbag_of_wordsen un RDD de LabeledPoints :
from pyspark.mllib.classification import LabeledPoint
dataframe.rdd.map(lambda row: LabeledPoint(row.label_index, row.bag_of_words.toArray()))Avec un label et des données, unLabeledPointcontient tout ce qui est nécessaire à un classifieur pour réaliser son apprentissage.
