lines = sc.textFile("../data/ml-100k/u.data")
print(lines.first())
# user_id  movie_id  rating  timestamp  

"""
Explanation: Feature Extraction from the MovieLens 100k Dataset
MovieLens Dataset (https://grouplens.org/datasets/movielens/100k/) 
100'000 movie ratings by 1'000 users for 1'700 different movies.
End of explanation
"""


from pyspark.mllib.recommendation import Rating
from pyspark.sql import Row
ratings_split = lines.map(lambda s: s.split("\t"))
ratings = ratings_split.map(
    lambda col: Row(userid=int(col[0]), movieid=int(col[1]), rating=float(col[2]))
)

"""
Explanation: Read text file line by line. Split it at '\t', parse it and create Row objects with ratings.
End of explanation
"""


movies = sc.textFile("../data/ml-100k/u.item")
titles = movies.map(lambda s: s.split('|')).map(lambda line: (int(line[0]), line[1])).collectAsMap()

"""
Explanation: Loead titles and movie IDs. Split this text file at |. Column 0 is the ID, column 1 is the title of the movie.
End of explanation
"""


first10 = ratings.map(lambda r: '{:3d}'.format(r.userid) + ": "+str(r.rating) + "  "+titles[r.movieid]).take(10)
for s in first10: print(s)

"""
Explanation: Show first ten ratings.
End of explanation
"""


from pyspark.ml.recommendation import ALS

"""
Explanation: Training of the ALS Model
End of explanation
"""


als = ALS(rank=10, maxIter=10, regParam=0.01, userCol="userid", 
          itemCol="movieid", ratingCol="rating")

"""
Explanation: Matrix factorization by ALS (Alternate Least Squares) using 10 iterationens and rank 10, that is, 10 latent dimensions. Use 0.01 as regularization parameter.
End of explanation
"""


ratingsDF = spark.createDataFrame(ratings)
model = als.fit(ratingsDF)

"""
Explanation: Train model (fit).
End of explanation
"""


model.userFactors.first()

"""
Explanation: User factor matrix: Latent features of user 10.
End of explanation
"""


model.itemFactors.first()

"""
Explanation: item (movie) factor matrix: Latent features of movie with ID 10.
End of explanation
"""


user = 51
user_ratings = ratings.filter(lambda r: r["userid"]==user).map(
    lambda r: str(r["rating"]) + ": " + titles[r["movieid"]]).collect()
for ur in user_ratings: print(ur)

"""
Explanation: User-specific Movie Recommendations
Ratings provided by user 51.
End of explanation
"""


movies = [56, 176, 161, 179]
user_movies = spark.createDataFrame([(user, m) for m in movies], 
                                    ["userid", "movieid"])
predictions = model.transform(user_movies).rdd.map(
    lambda r: "User " +str(r["userid"]) + ": predicted rating " + 
        "{:5.3f}".format(r["prediction"]) + 
        " for " + titles[r["movieid"]]).collect()
for p in predictions: print(p)

"""
Explanation: Predicted rating for this user for four arbitrarily chosen  movies.
End of explanation
"""


movie = 50
print(titles[movie])

movie_features = model.itemFactors.rdd
import numpy as np
query_movie = np.array(movie_features.filter(lambda r: r.id == movie).map(lambda r: r.features).first())
print(query_movie)

def cosine_similarity(a, b):
    return np.dot(a, b)/(np.linalg.norm(a)*np.linalg.norm(b))

"""
Explanation: Comparison of similar movies using latent factors
End of explanation
"""


top10 = movie_features.map(lambda r: (r.id, cosine_similarity(query_movie, r.features))
                  ).sortBy(lambda r: -r[1]).map(lambda r: (titles[r[0]], r[1])).take(10)
for t, r in top10: print("{:.3}".format(r)+": "+t)

"""
Explanation: Top-10 of the most similar movies to "Star Wars".
End of explanation
"""


