机器学习:spark ALS预测电影-python

Python代码:

1. #!/usr/bin/env python  
2. # coding=utf-8  
3. '''  
4. 运行命令/yourpath/spark/bin/spark-submit --driver-memory 1g MovieLensALS.py movieLensDataDir personalRatingsFile  
5. movieLensDataDir 电影评分数据集目录 比如 ml-1m/  
6. personalRatingsFile 需要推荐的某用户的评价数据 格式参考ratings.dat  
7.   
8. '''  
9. import sys  
10. import itertools  
11. from math import sqrt  
12. from operator import add  
13. from os.path import join, isfile, dirname  
14.   
15. from pyspark import SparkConf, SparkContext  
16. from pyspark.mllib.recommendation import ALS  
17.   
18. def parseRating(line):  
19.     """  
20.         Parses a rating record in MovieLens format userId::movieId::rating::timestamp .  
21.     """  
22.     fields = line.strip().split("::")  
23.     return long(fields[3]) % 10, (int(fields[0]), int(fields[1]), float(fields[2]))  
24.   
25. def parseMovie(line):  
26.     """  
27.         Parses a movie record in MovieLens format movieId::movieTitle .  
28.     """  
29.     fields = line.strip().split("::")  
30.     return int(fields[0]), fields[1]  
31.   
32. def loadRatings(ratingsFile):  
33.     """  
34.        Load ratings from file.  
35.     """  
36.     if not isfile(ratingsFile):  
37.         print "File %s does not exist." % ratingsFile  
38.         sys.exit(1)  
39.     f = open(ratingsFile, 'r')  
40.     ratings = filter(lambda r: r[2] > 0, [parseRating(line)[1] for line in f])  
41.     f.close()  
42.     if not ratings:  
43.         print "No ratings provided."  
44.         sys.exit(1)  
45.     else:  
46.         return ratings  
47.   
48. def computeRmse(model, data, n):  
49.     """  
50.         Compute RMSE (Root Mean Squared Error).  
51.     """  
52.     predictions = model.predictAll(data.map(lambda x: (x[0], x[1])))  
53.     predictionsAndRatings = predictions.map(lambda x: ((x[0], x[1]), x[2])).join(data.map(lambda x: ((x[0], x[1]), x[2]))).values()  
54.     return sqrt(predictionsAndRatings.map(lambda x: (x[0] - x[1]) ** 2).reduce(add) / float(n))  
55.   
56. if __name__ == "__main__":  
57.     if (len(sys.argv) != 3):  
58.         print "Usage: /path/to/spark/bin/spark-submit --driver-memory 1g MovieLensALS.py movieLensDataDir personalRatingsFile"  
59.         sys.exit(1)  
60.   
61.     # set up environment  
62.     conf = SparkConf().setAppName("MovieLensALS").set("spark.executor.memory", "1g")  
63.     sc = SparkContext(conf=conf)  
64.   
65.     # load personal ratings  
66.     myRatings = loadRatings(sys.argv[2])  
67.     myRatingsRDD = sc.parallelize(myRatings, 1)  
68.   
69.     movieLensHomeDir = sys.argv[1]  
70.   
71.     # ratings is an RDD of (last digit of timestamp, (userId, movieId, rating))  
72.     ratings = sc.textFile(join(movieLensHomeDir, "ratings.dat")).map(parseRating)  
73.   
74.     #  movies is an RDD of (movieId, movieTitle)  
75.     movies = dict(sc.textFile(join(movieLensHomeDir, "movies.dat")).map(parseMovie).collect())  
76.   
77.     numRatings = ratings.count()  
78.     numUsers = ratings.values().map(lambda r: r[0]).distinct().count()  
79.     numMovies = ratings.values().map(lambda r: r[1]).distinct().count()  
80.       
81.     myRatedMovieIds = set([x[1] for x in myRatings])  
82.   
83.     print "Got %d ratings from %d users on %d movies." % (numRatings, numUsers, numMovies)  
84.   
85.      
86.     # split ratings into train , validation   
87.     # last digit of the timestamp, add myRatings to train, and cache them  
88.   
89.     # training, validation, test are all RDDs of (userId, movieId, rating)  
90.   
91.     numPartitions = 4  
92.     #training = ratings.filter(lambda x: x[0] 8).values().union(myRatingsRDD).repartition(numPartitions).cache()  
93.   
94.     validation = ratings.filter(lambda x: x[0] >= 8 and x[0] 10).values().repartition(numPartitions).cache()  
95.   
96.   
97.       
98.   
99.   
100.     numTraining = training.count()  
101.     numValidation = validation.count()  
102.   
103.     print "Training: %d, validation: %d" % (numTraining, numValidation)  
104.   
105.     # train models and evaluate them on the validation set  
106.   
107.     ranks = [10,12]  
108.     lambdas = [0.01,0.4,1.0]  
109.     numIters = [10]  
110.     bestModel = None  
111.     bestValidationRmse = float("inf")  
112.     bestRank = 0  
113.     bestLambda = -1.0  
114.     bestNumIter = -1  
115.   
116.     for rank, lmbda, numIter in itertools.product(ranks, lambdas, numIters):  
117.         model = ALS.train(training, rank, numIter, lmbda)  
118.         validationRmse = computeRmse(model, validation, numValidation)  
119.         print "RMSE (validation) = %f for the model trained with " % validationRmse + "rank = %d, lambda = %.4f, and numIter = %d." % (rank, lmbda, numIter)  
120.         if (validationRmse 
121.             bestModel = model  
122.             bestValidationRmse = validationRmse  
123.             bestRank = rank  
124.             bestLambda = lmbda  
125.             bestNumIter = numIter  
126.   
127.   
128.   
129.     # evaluate the best model on the test set  
130.     print "The best model was trained with rank = %d and lambda = %.4f, and numIter = %d ,and Rmse %.4f" % (bestRank, bestLambda,bestNumIter,bestValidationRmse)  
131.       
132.     #exit()  
133.     #通过计算得到rank = 10   lambda = 0.45  numIter = 20 结果最好  
134.   
135.     bestModel =  ALS.train(training, 10, 20, 0.45);  
136.     # training, validation, test are all RDDs of (userId, movieId, rating)  
137.   
138.     #make personalized recommendations  
139.     #排除该用户已评价过的电影  
140.      
141.     testdata = training.filter(lambda x: x[0] not in myRatedMovieIds).map(lambda p: (int(p[0]), int(p[1])))  
142.   
143.   
144.     predictions = bestModel.predictAll(testdata).map(lambda r: ((r[0], r[1]), r[2]))  
145.     #对预测结果按分值排序 取前5  
146.     recommendations = predictions.sortBy(lambda x:x[1],False).take(5)  
147.      
148.     print "Movies recommended for you:"  
149.     for i in xrange(len(recommendations)):  
150.         print ("%2d: %s %s" % (i + 1, recommendations[i][0],recommendations[i][1]))  
151.   
152.     # clean up  
153.     sc.stop() 
     

数据集采用MovieLens
代码参考https://github.com/databricks/spark-training/blob/master/machine-learning/python/solution/MovieLensALS.py

personalRatingsFile
0::1::?::1400000000::Toy Story (1995)
0::780::?::1400000000::Independence Day (a.k.a. ID4) (1996)
0::590::?::1400000000::Dances with Wolves (1990)
0::1210::?::1400000000::Star Wars: Episode VI - Return of the Jedi (1983)
0::648::?::1400000000::Mission: Impossible (1996)
0::344::?::1400000000::Ace Ventura: Pet Detective (1994)
0::165::?::1400000000::Die Hard: With a Vengeance (1995)
0::153::?::1400000000::Batman Forever (1995)
0::597::?::1400000000::Pretty Woman (1990)
0::1580::?::1400000000::Men in Black (1997)
0::231::?::1400000000::Dumb & Dumber (1994)