Principes de Big Data avec PySpark
Upendra Devisetty
Science Analyst, CyVerse
Le filtrage collaboratif repère des utilisateurs aux intérêts communs
Il est couramment utilisé pour les systèmes de recommandation
Approches de filtrage collaboratif :
Filtrage collaboratif Utilisateur-Utilisateur : trouve des utilisateurs semblables à l'utilisateur cible
Filtrage collaboratif Article-Article : trouve et recommande des articles semblables à ceux de l'utilisateur cible
La classe Rating est un enrobage autour du tuple (user, product, rating)
Utile pour analyser le RDD et créer un tuple user, product, rating
from pyspark.mllib.recommendation import Rating
r = Rating(user = 1, product = 2, rating = 5.0)
(r[0], r[1], r[2])
(1, 2, 5.0)
Scinder les données en ensembles d'entraînement et de test est essentiel pour évaluer un modèle prédictif
En général, une plus grande part va à l'entraînement qu'au test
La méthode randomSplit() de PySpark scinde aléatoirement selon les pondérations et renvoie plusieurs RDD
data = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
training, test=data.randomSplit([0.6, 0.4])
training.collect()
test.collect()
[1, 2, 5, 6, 9, 10]
[3, 4, 7, 8]
L'algorithme Alternating Least Squares (ALS) de spark.mllib offre le filtrage collaboratif
ALS.train(ratings, rank, iterations)
r1 = Rating(1, 1, 1.0)
r2 = Rating(1, 2, 2.0)
r3 = Rating(2, 1, 2.0)
ratings = sc.parallelize([r1, r2, r3])
ratings.collect()
[Rating(user=1, product=1, rating=1.0),
Rating(user=1, product=2, rating=2.0),
Rating(user=2, product=1, rating=2.0)]
model = ALS.train(ratings, rank=10, iterations=10)
La méthode predictAll() retourne une liste de notes prédites pour des paires utilisateur-produit en entrée
Elle prend un RDD sans notes pour générer les prédictions
unrated_RDD = sc.parallelize([(1, 2), (1, 1)])
predictions = model.predictAll(unrated_RDD)
predictions.collect()
[Rating(user=1, product=1, rating=1.0000278574351853),
Rating(user=1, product=2, rating=1.9890355703778122)]
rates = ratings.map(lambda x: ((x[0], x[1]), x[2]))
rates.collect()
[((1, 1), 1.0), ((1, 2), 2.0), ((2, 1), 2.0)]
preds = predictions.map(lambda x: ((x[0], x[1]), x[2])) preds.collect()[((1, 1), 1.000027857), ((1, 2), 1.9890355703)]
rates_preds = rates.join(preds)
rates_preds.collect()
[((1, 2), (2.0, 1.9890355703)), ((1, 1), (1.0, 1.000027857))]
La MSE est la moyenne du carré de (actual rating - predicted rating)
MSE = rates_preds.map(lambda r: (r[1][0] - r[1][1])**2).mean()
Principes de Big Data avec PySpark