OopsOutOfMemory 盛利's Blog
专注大数据领域,分布式计算,Spark Contributor
<原创文章>

Spark机器学习库mllib之协同过滤

盛利的博客
, in 19 November 2014

很久就想写一篇ML的实践文章,虽然看过肯多资料,总觉得纸上谈兵印象不深刻,过不了多久就忘了,现在就借Spark的Mllib来简单的实际一下推荐算法吧。

说起推荐算法,大家耳熟能详的就是CF(协同过滤),这次就拿CF中ALS(alternating least squares),交替最小二乘,来做个例子吧。 CF里面的算法比较多,有基于物品的,基于用户的,ALS是基于矩阵分解的,关于对推荐算法的小结,请参考我的推荐算法总结Recommendation

Mllib

先介绍下mllib,mllib是运行在Spark上一个机器学习算法库。借助Spark的内存计算,可以使机器学习的模型计算时间大大缩短。 目前,spark1.0.0中的mllib中已经有很多算法了,具体可以参见官方网站http://spark.apache.org/docs/latest/mllib-guide.html

我们知道,协同过滤是基于用户行为的一种推荐算法,需要用户对Item的评价。 于是乎我们还是找到最经典的数据集movielens,地址http://grouplens.org/datasets/movielens/

Down下来ml-100k,解压后有很多文件,可以看README里面对数据集的介绍。

user id | item id | rating | timestamp                
1       1      5       874965758  
1       2      3       876893171  
1       3      4       878542960  
1       4      3       876893119  
1       5      3       889751712  
1       7      4       875071561  
1       8      1       875072484  
1       9      5       878543541  
2       1      2       875072262  
2       3      5       875071805  
2       5      5       875071608  
2       6      5       878543541  
2       8      4       887432020  
2       9      5       875071515  
2       1      1       878542772  
2       2      4       875072404  

这里有user对某个movie的评分rating和时间timstamp

先预处理一下数据

cat u1.base | awk -F "\t" '{print $1"::"$2"::"$3"::"$4}' > ratings.dat  
cat u.item | awk -F "|" '{print $1"\t"$2"\t"$3}' > movies.dat  
数据结果

```scala
user id::movie id::rating:: timestamp  
1::1::5::874965758  
1::5::3::889751712  
1::7::4::875071561  
1::8::1::875072484  
1::9::5::878543541  
2::258::3::888549961  
2::269::4::888550774  
2::272::5::888979061  
2::273::4::888551647  
2::274::3::888551497  

movie id ::movie id :: movie release date  
1::Toy Story (1995)::01-Jan-1995  
2::GoldenEye (1995)::01-Jan-1995  
3::Four Rooms (1995)::01-Jan-1995  
4::Get Shorty (1995)::01-Jan-1995  
5::Copycat (1995)::01-Jan-1995  
6::Shanghai Triad (Yao a yao yao dao waipo qiao) (1995)::01-Jan-1995  
7::Twelve Monkeys (1995)::01-Jan-1995  
8::Babe (1995)::01-Jan-1995  
9::Dead Man Walking (1995)::01-Jan-1995  
10::Richard III (1995)::22-Jan-1996  
11::Seven (Se7en) (1995)::01-Jan-1995  
12::Usual Suspects, The (1995)::14-Aug-1995  

OK,下面我们要用官方的ALS算法例子来运行下这个推荐。 首先导入mllib包,我们需要用到ALS算法类和Rating评分类

import org.apache.spark.mllib.recommendation.ALS  
import org.apache.spark.mllib.recommendation.Rating  

//加载数据  
val data = sc.textFile("/app/hadoop/ml-100k/ratings.dat")  
//data中每条数据经过map的split后会是一个数组,模式匹配后,会new一个Rating对象  
val ratings = data.map(_.split("::") match { case Array(user, item, rate, ts) =>  
    Rating(user.toInt, item.toInt, rate.toDouble)  
  })  

最终会new成对象

```scala
scala> ratings take 2  
......  
14/06/25 17:51:06 INFO scheduler.DAGScheduler: Computing the requested partition locally  
14/06/25 17:51:06 INFO rdd.HadoopRDD: Input split: file:/app/hadoop/ml-100k/ratings.dat:0+1826544  
14/06/25 17:51:07 INFO spark.SparkContext: Job finished: take at <console>:22, took 0.062239021 s  
res0: Array[org.apache.spark.mllib.recommendation.Rating] = Array(Rating(1,1,5.0), Rating(1,2,3.0))  
//设置潜在因子个数为10  
scala> val rank = 10  
rank: Int = 10  
//要迭代计算30次  
scala> val numIterations = 30  
numIterations: Int = 30  

接下来调用ALS.train()方法,进行模型训练:

val model = ALS.train(ratings, rank, numIterations, 0.01)  
14/06/25 17:53:04 INFO storage.MemoryStore: ensureFreeSpace(200) called with curMem=84002, maxMem=308713881  
14/06/25 17:53:04 INFO storage.MemoryStore: Block broadcast_60 stored as values to memory (estimated size 200.0 B, free 294.3 MB)  
model: org.apache.spark.mllib.recommendation.MatrixFactorizationModel = org.apache.spark.mllib.recommendation.MatrixFactorizationModel@17596ee0  

训练完后,我们要对比一下预测的结果,我们那训练集当作测试集,来进行对比测试:

scala> val usersProducts = ratings.map { case Rating(user, product, rate) =>  
     |   (user, product)  
     | }  
usersProducts: org.apache.spark.rdd.RDD[(Int, Int)] = MappedRDD[623] at map at <console>:21  
//预测后的用户,电影,评分  
scala> val predictions =   
     |   model.predict(usersProducts).map { case Rating(user, product, rate) =>   
     |     ((user, product), rate)  
     |   }  
predictions: org.apache.spark.rdd.RDD[((Int, Int), Double)] = MappedRDD[632] at map at <console>:30  

我们用均方根误差来评价一个模型的好坏,所以我们要算一下MSE,来判定这个模型的准确率,其值越小说明越准确。

join一下,然后再计算:

//原始{(用户,电影),评分} join  预测后的{(用户,电影),评分}  
val ratesAndPreds = ratings.map { case Rating(user, product, rate) =>   
  ((user, product), rate)  
}.join(predictions)  
ratesAndPreds.collect take 3  
14/06/25 17:59:35 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 632.0, whose tasks have all completed, from pool   
14/06/25 17:59:35 INFO scheduler.DAGScheduler: Stage 632 (collect at <console>:34) finished in 1.906 s  
14/06/25 17:59:35 INFO spark.SparkContext: Job finished: collect at <console>:34, took 1.939437725 s  
res11: Array[((Int, Int), (Double, Double))] = Array(((933,627),(2.0,1.6977799770529198)), ((537,24),(1.0,2.3191609228008327)), ((717,125),(4.0,3.795616142104737)))  

join后的结果,就是每个用户对电影的实际打分和预测打分的一个对比,例如:

(用户电影)(原始评分预测的评分
(933,627)(2.01.6977799770529198)
(537,24),(1.0, 2.3191609228008327)
(717,125),(4.0,3.795616142104737)
 ......

最后计算均方根误差:

val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =>   
 val err = (r1 - r2)  
 err * err  
.mean()  
14/06/25 18:02:28 INFO scheduler.TaskSetManager: Finished TID 79 in 554 ms on localhost (progress: 1/1)  
14/06/25 18:02:28 INFO scheduler.DAGScheduler: Stage 702 (mean at <console>:36) finished in 0.556 s  
14/06/25 18:02:28 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 702.0, whose tasks have all completed, from pool   
14/06/25 18:02:28 INFO spark.SparkContext: Job finished: mean at <console>:36, took 0.585592521 s  
MSE: Double = 0.4254804561682655  

顺便提一下预测的API有三个重载,上面用的是第二个: 调用model的API predict

scala> model.predict  

def predict(user: Int, product: Int): Double                                                                                                 
def predict(usersProducts: org.apache.spark.rdd.RDD[(Int, Int)]): org.apache.spark.rdd.RDD[org.apache.spark.mllib.recommendation.Rating]     
def predict(usersProductsJRDD: org.apache.spark.api.java.JavaRDD[Array[Byte]]): org.apache.spark.api.java.JavaRDD[Array[Byte]]          

我们也可以传入user id, product id 来与此 某个用户 对某个 电影 的评分

model.predict(1,2)  
14/06/25 18:17:36 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 963.0, whose tasks have all completed, from pool   
14/06/25 18:17:36 INFO scheduler.DAGScheduler: Stage 963 (lookup at MatrixFactorizationModel.scala:46) finished in 0.035 s  
14/06/25 18:17:36 INFO spark.SparkContext: Job finished: lookup at MatrixFactorizationModel.scala:46, took 0.066978561 s  
res13: Double = 2.9204503120927363  

模型都有了,推荐系统怎么设计就根据实际需求了 :)

总结:

MLlib充分利用了Spark的快速内存计算,迭代效率高的优势,将机器学习的模型计算性能提到另一片天地,这也就是为什么最近Spark备受推崇,那么火的原因。 目前Mllib的算法库还不是很多,但是Mahout都宣布不接受Mapreduce算法库,都迁移到spark上来了,看来未来机器学习要靠Spark了。至于为什么对于协同过滤先支持的是ALS,也是看中的ALS算法的并行度比较好,在Spark上更能发挥该算法的优势吧。

——EOF——

(The End)
<原创文章> From OopsOutOfMemory 盛利's Blog
转载请注明出自: http://oopsoutofmemory.github.io/machine%20learning/spark/mllib/2014/11/19/spark-ji-qi-xue-xi-ku-mllib-zhi-xie-tong-guo-lv