sc

# Chargement des fichiers si ce n'est déjà fait
#Renseignez ici le dossier où vous souhaitez stocker le fichier téléchargé.
DATA_PATH="" 
import urllib.request
# fichier réduit
f = urllib.request.urlretrieve("http://www.math.univ-toulouse.fr/~besse/Wikistat/data/ml-ratings100k.csv",DATA_PATH+"ml-ratings100k.csv")

"""
Explanation: Ateliers: Technologies des grosses données
Recommandation de Films par Filtrage Collaboratif: NMF de la librairie SparkML de <a href="http://spark.apache.org/"><img src="http://spark.apache.org/images/spark-logo-trademark.png" style="max-width: 100px; display: inline" alt="Spark"/></a>
1. Introduction
Ce calepin traite d'un problème classique de recommandation par filtrage collaboratif en utilisant les ressources de la librairie MLlib de Spark avec l'API pyspark. Le problème général est décrit en introduction et dans une vignette de Wikistat. Il est appliqué aux données publiques du site GroupLens. L'objectif est de tester les méthodes et la procédure d'optimisation sur le plus petit jeu de données composé de 100k notes  de 943 clients sur 1682 films où chaque client a au moins noté 20 films. Les jeux de données plus gros (1M, 10M, 20M notes) peuvent être utilisés pour "passer à l'échelle volume". 
Ce calepin s'inspire des exemples de la documentation et d'un tutoriel de Jose A. Dianes. Le sujet a été traité lors d'un Spark Summit.
L'objectif est d'utiliser ces seules données pour proposer des recommandations.  Les données initiales sont sous la forme d'une matrice très creuse (sparse) contenant des notes ou évaluations. Attention, les "0" de la matrice ne sont pas des notes mais des données manquantes, le film n'a pas encore été vu ou évalué. 
Un algorithme satisfaisant à l'objectif de complétion de grande matrice creuse, et implémenté dans un logiciel libre d'accès est disponible dans la librairie softImpute de R. SOn utilisaiton est décrite dans un autre calepin. La version de NMF de MLlib de Spark autorise permet également la complétion.
En revanche,la version  de NMF incluse dans la librairie Scikit-learn traite également des matrices creuses mais le critère (moindres carrés) optimisé considère les "0" comme des notes nulles, pas comme des données manquantes. Elle n'est pas adaptée au problème de complétion, contrairement à celle de MLliB. Il faudrait sans doute utiliser la librairie nonnegfac en Python  de Kim et al. (2014); à tester!
Dans la première partie, le plus petit fichier est partagé en trois échantillons: apprentissage, validation et test; l'optimisation du rang de la factorisation (nombre de facteurs latents) est réalisée par minimisation de l'erreur estimée sur l'échantillon de validation.
Ensuite le plus gros fichier est utilisé pour évaluer l'impact de la taille de la base d'apprentissage.
2 Importation des données en HDFS
Les données doivent être stockées à un emplacement accessibles de tous les noeuds du cluster pour permettre la construction de la base de données réparties (RDD). Dans une utilisation monoposte (standalone) de Spark, elles sont simplement chargées dans le répertoire courant.
End of explanation
"""


# Importer les données au format texte dans un RDD

small_ratings_raw_data = sc.textFile(DATA_PATH+"ml-ratings100k.csv")

# Identifier et afficher la première ligne
small_ratings_raw_data_header = small_ratings_raw_data.take(1)[0]
print(small_ratings_raw_data_header)

# Create RDD without header
all_lines = small_ratings_raw_data.filter(lambda l : l!=small_ratings_raw_data_header)

# Séparer les champs (user, item, note) dans un nouveau RDD
from pyspark.sql import Row
split_lines = all_lines.map(lambda l : l.split(","))
ratingsRDD = split_lines.map(lambda p: Row(user=int(p[0]), item=int(p[1]),
                                     rating=float(p[2]), timestamp=int(p[3])))

# .cache() : le RDD est conservé en mémoire une fois traité
ratingsRDD.cache()

# Display the two first rows
ratingsRDD.take(2)

# Convert RDD to DataFrame
ratingsDF = spark.createDataFrame(ratingsRDD)
ratingsDF.take(2)

"""
Explanation: Les données sont lues comme une seule ligne de texte avant d'être restructurées au bon format d'une matrice creuse à savoir une liste de triplets contenant les  indices de ligne, de colonne et la note pour les seules valeurs renseignées.
End of explanation
"""


tauxTrain=0.6
tauxVal=0.2
tauxTes=0.2
# Si le total est inférieur à 1, les données sont sous-échantillonnées.
(trainDF, validDF, testDF) = ratingsDF.randomSplit([tauxTrain, tauxVal, tauxTes])
# validation et test à prédire, sans les notes
validDF_P = validDF.select("user", "item")
testDF_P = testDF.select("user", "item")

trainDF.take(2), validDF_P.take(2), testDF_P.take(2)

"""
Explanation: 3. Optimisation du rang sur l'échantillon 10k
Le fichier comporte 10 000 évaluations croisant les avis de mille utilisateurs sur les films qu'ils ont vus parmi 1700.
3.1 Constitution des échantillons
Séparation aléatoire en trois échantillons apprentissage, validation et test. Le paramètre de rang est optimisé en minimisant l'estimaiton de l'erreur sur l'échantillon test. Cette stratégie, plutôt qu'ue validation croisée est plus adaptée à des données massives.
End of explanation
"""


from pyspark.ml.recommendation import ALS
import math
import collections
# Initialisation du générateur
seed = 5
# Nombre max d'itérations (ALS)
maxIter = 10
# Régularisation L1; à optimiser également
regularization_parameter = 0.1
# Choix d'une grille pour les valeurs du rang à optimiser
ranks = [4, 8, 12]

#Initialisation variable 
# création d'un dictionaire pour stocker l'erreur par rang testé
errors = collections.defaultdict(float)
tolerance = 0.02
min_error = float('inf')
best_rank = -1
best_iteration = -1

from pyspark.ml.evaluation import RegressionEvaluator

for rank in ranks:
    als = ALS( rank=rank, seed=seed, maxIter=maxIter,
                      regParam=regularization_parameter)
    model = als.fit(trainDF)
    # Prévision de l'échantillon de validation
    predDF = model.transform(validDF).select("prediction","rating")
    #Remove unpredicter row due to no-presence of user in the train dataset
    pred_without_naDF = predDF.na.drop()
    # Calcul du RMSE
    evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                                predictionCol="prediction")
    rmse = evaluator.evaluate(pred_without_naDF)
    print("Root-mean-square error for rank %d = "%rank + str(rmse))
    errors[rank] = rmse
    if rmse < min_error:
        min_error = rmse
        best_rank = rank
# Meilleure solution
print('Rang optimal: %s' % best_rank)

"""
Explanation: 3.2 Optimisation du rang de la NMF
L'erreur d'imputation des données, donc de recommandation, est estimée sur l'échantillon de validation pour différentes valeurs (grille) du rang de la factorisation matricielle. 
Il faudrait en principe aussi optimiser la valeur du paramètre de pénalisation pris à 0.1 par défaut.
Point important: l'erreur d'ajustement de la factorisation ne prend en compte que les valeurs listées dans la matrice creuses, pas les "0" qui sont des données manquantes.
End of explanation
"""


# Quelques prévisions
pred_without_naDF.take(3)

"""
Explanation: 3.3 Résultats et test
End of explanation
"""


#On concatane la DataFrame Train et Validatin
trainValidDF = trainDF.union(validDF)
# On crée un model avec le nouveau Dataframe complété d'apprentissage et le rank fixé à la valeur optimal 
als = ALS( rank=best_rank, seed=seed, maxIter=maxIter,
                  regParam=regularization_parameter)
model = als.fit(trainValidDF)
#Prediction sur la DataFrame Test
testDF = model.transform(testDF).select("prediction","rating")
#Remove unpredicter row due to no-presence of user in the trai dataset
pred_without_naDF = predDF.na.drop()
# Calcul du RMSE
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                            predictionCol="prediction")
rmse = evaluator.evaluate(pred_without_naDF)
print("Root-mean-square error for rank %d = "%best_rank + str(rmse))



"""
Explanation: Prévision finale de l'échantillon test.
End of explanation
"""


# Chargement des fichiers si ce n'est déjà fait
import urllib.request
# fichier complet mais compressé
f = urllib.request.urlretrieve("http://www.math.univ-toulouse.fr/~besse/Wikistat/data/ml-ratings20M.zip",DATA_PATH+"ml-ratings20M.zip")

#Unzip downloaded file
import zipfile
zip_ref = zipfile.ZipFile(DATA_PATH+"ml-ratings20M.zip", 'r')
zip_ref.extractall(DATA_PATH)
zip_ref.close()

# Importer les données au format texte dans un RDD
ratings_raw_data = sc.textFile(DATA_PATH+"ratings20M.csv")
# Identifier et afficher la première ligne
ratings_raw_data_header = ratings_raw_data.take(1)[0]
ratings_raw_data_header

# Create RDD without header
all_lines = ratings_raw_data.filter(lambda l : l!=ratings_raw_data_header)


# Séparer les champs (user, item, note) dans un nouveau RDD
split_lines = all_lines.map(lambda l : l.split(","))
ratingsRDD = split_lines.map(lambda p: Row(user=int(p[0]), item=int(p[1]),
                                     rating=float(p[2]), timestamp=int(p[3])))

# Display the two first rows
ratingsRDD.take(2)

# Convert RDD to DataFrame
ratingsDF = spark.createDataFrame(ratingsRDD)
ratingsDF.take(2)

"""
Explanation: 3 Analyse du fichier complet
MovieLens propose un plus gros fichier avec 20M de notes (138000 utilisateurs, 27000 films). Ce fichier est utilisé pour extraire un fichier test de deux millions de notes à reconstruire. Les paramètres précédemment optimisés, ils pourraient sans doute l'être mieux, sont appliqués pour une succesion d'estimation / prévision avec une taille croissante de l'échantillon d'apprentissage. Il aurait été plus élégant d'automatiser le travail dans une boucle mais lorsque les données sont les plus volumineuses des comportement mal contrôlés de Spark peuvent provoquer des plantages par défaut de mémoire.
3.1 Lecture des données
Le fichier est prétraité de manière analogue.
End of explanation
"""


tauxTest=0.1
# Si le total est inférieur à 1, les données sont sous-échantillonnées.
(trainTotDF,  testDF) = ratingsDF.randomSplit([1-tauxTest, tauxTest])

# Sous-échantillonnage de l'apprentissage permettant de 
# tester pour des tailles croissantes de cet échantillon
tauxEch=0.2
(trainDF, DropData) = trainTotDF.randomSplit([tauxEch, 1-tauxEch])

testDF.take(2), trainDF.take(2)

"""
Explanation: 3.2 Echantillonnage
Extraction de l'échantillon test et éventuellement sous-échantillonnage de l'échantillon d'apprentissage.
End of explanation
"""


import time
time_start=time.time()
# Initialisation du générateur
seed = 5
# Nombre max d'itérations (ALS)
maxIter = 10
# Régularisation L1 (valeur par défaut)
regularization_parameter = 0.1
best_rank = 8
# Estimation pour chaque valeur de rang
als = ALS(rank=rank, seed=seed, maxIter=maxIter,
                      regParam=regularization_parameter)
model = als.fit(trainDF)
time_end=time.time()
time_als=(time_end - time_start)
print("ALS prend %d s" %(time_als)) 

"""
Explanation: 3.3 Estimation du modèle
Le modèle est estimé en utilisant les valeurs des paramètres obtenues dans l'étape précédente.
End of explanation
"""


# Prévision de l'échantillon de validation
predDF = model.transform(testDF).select("prediction","rating")
#Remove unpredicter row due to no-presence of user in the train dataset
pred_without_naDF = predDF.na.drop()
# Calcul du RMSE
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                            predictionCol="prediction")
rmse = evaluator.evaluate(pred_without_naDF)
print("Root-mean-square error for rank %d = "%best_rank + str(rmse))

trainDF.count()

"""
Explanation: 3.4 Prévision de l'échantillon test et erreur
End of explanation
"""


