用Spark学习矩阵分解推荐算法

在矩阵分解在协同过滤推荐算法中的应用中，咱们对矩阵分解在推荐算法中的应用原理作了总结，这里咱们就从实践的角度来用Spark学习矩阵分解推荐算法。

1、Spark推荐算法概述

　　　　在Spark MLlib中，推荐算法这块只实现了基于矩阵分解的协同过滤推荐算法。而基于的算法是FunkSVD算法，即将m个用户和n个物品对应的评分矩阵M分解为两个低维的矩阵：
\[ M_{m \times n}=P_{m \times k}^TQ_{k \times n} \]

　　　　其中k为分解成低维的维数，通常远比m和n小。若是你们对FunkSVD算法不熟悉，能够复习对应的原理篇。

2、2. Spark推荐算法类库介绍

　　　　在Spark MLlib中，实现的FunkSVD算法支持Python,Java,Scala和R的接口。因为前面的实践篇咱们都是基于Python，本文的后面的介绍和使用也会使用MLlib的Python接口。

　　　　Spark MLlib推荐算法python对应的接口都在pyspark.mllib.recommendation包中，这个包有三个类，Rating, MatrixFactorizationModel和ALS。虽然里面有三个类，可是算法只是FunkSVD算法。下面介绍这三个类的用途。

　　　　Rating类比较简单，仅仅只是为了封装用户，物品与评分这3个值。也就是说，Rating类里面只有用户，物品与评分三元组， 并无什么函数接口。

　　　　ALS负责训练咱们的FunkSVD模型。之因此这儿用交替最小二乘法ALS表示，是由于Spark在FunkSVD的矩阵分解的目标函数优化时，使用的是ALS。ALS函数有两个函数，一个是train,这个函数直接使用咱们的评分矩阵来训练数据，而另外一个函数trainImplicit则稍微复杂一点，它使用隐式反馈数据来训练模型，和train函数相比，它多了一个指定隐式反馈信心阈值的参数，好比咱们能够将评分矩阵转化为反馈数据矩阵，将对应的评分值根据必定的反馈原则转化为信心权重值。因为隐式反馈原则通常要根据具体的问题和数据来定，本文后面只讨论普通的评分矩阵分解。

　　　　MatrixFactorizationModel类是咱们用ALS类训练出来的模型，这个模型能够帮助咱们作预测。经常使用的预测有某一用户和某一物品对应的评分，某用户最喜欢的N个物品，某物品可能会被最喜欢的N个用户，全部用户各自最喜欢的N物品，以及全部物品被最喜欢的N个用户。

　　　　对于这些类的用法咱们再后面会有例子讲解。

3、Spark推荐算法重要类参数

　　　　这里咱们再对ALS训练模型时的重要参数作一个总结。　

　　　　1) ratings : 评分矩阵对应的RDD。须要咱们输入。若是是隐式反馈，则是评分矩阵对应的隐式反馈矩阵。

　　　　2) rank : 矩阵分解时对应的低维的维数。即\(P_{m \times k}^TQ_{k \times n}\)中的维度k。这个值会影响矩阵分解的性能，越大则算法运行的时间和占用的内存可能会越多。一般须要进行调参，通常能够取10-200之间的数。

　　　　3) iterations :在矩阵分解用交替最小二乘法求解时，进行迭代的最大次数。这个值取决于评分矩阵的维度，以及评分矩阵的系数程度。通常来讲，不须要太大，好比5-20次便可。默认值是5。

　　　　4) lambda: 在 python接口中使用的是lambda_,缘由是lambda是Python的保留字。这个值即为FunkSVD分解时对应的正则化系数。主要用于控制模型的拟合程度，加强模型泛化能力。取值越大，则正则化惩罚越强。大型推荐系统通常须要调参获得合适的值。

　　　　5) alpha : 这个参数仅仅在使用隐式反馈trainImplicit时有用。指定了隐式反馈信心阈值，这个值越大则越认为用户和他没有评分的物品之间没有关联。通常须要调参获得合适值。

　　　　从上面的描述能够看出，使用ALS算法仍是蛮简单的，须要注意调参的参数主要的是矩阵分解的维数rank, 正则化超参数lambda。若是是隐式反馈，还须要调参隐式反馈信心阈值alpha 。

4、4. Spark推荐算法实例

　　　　下面咱们用一个具体的例子来说述Spark矩阵分解推荐算法的使用。

　　　　完整代码参见个人github: https://github.com/nickchen121/machinelearning/blob/master/classic-machine-learning/matrix_factorization.ipynb

　　　　这里咱们使用MovieLens 100K的数据，数据下载连接在这。

　　　　将数据解压后，咱们只使用其中的u.data文件中的评分数据。这个数据集每行有4列，分别对应用户ID，物品ID，评分和时间戳。因为个人机器比较破，在下面的例子中，我只使用了前100条数据。所以若是你使用了全部的数据，后面的预测结果会与个人不一样。

　　　　首先须要要确保你安装好了Hadoop和Spark（版本不小于1.6），并设置好了环境变量。通常咱们都是在ipython notebook(jupyter notebook)中学习，因此最好把基于notebook的Spark环境搭好。固然不搭notebook的Spark环境也没有关系，只是每次须要在运行前设置环境变量。

　　　　若是你没有搭notebook的Spark环境，则须要先跑下面这段代码。固然，若是你已经搭好了，则下面这段代码不用跑了。

import os
import sys

#下面这些目录都是你本身机器的Spark安装目录和Java安装目录
os.environ['SPARK_HOME'] = "C:/Tools/spark-1.6.1-bin-hadoop2.6/"

sys.path.append("C:/Tools/spark-1.6.1-bin-hadoop2.6/bin")
sys.path.append("C:/Tools/spark-1.6.1-bin-hadoop2.6/python")
sys.path.append("C:/Tools/spark-1.6.1-bin-hadoop2.6/python/pyspark")
sys.path.append("C:/Tools/spark-1.6.1-bin-hadoop2.6/python/lib")
sys.path.append("C:/Tools/spark-1.6.1-bin-hadoop2.6/python/lib/pyspark.zip")
sys.path.append("C:/Tools/spark-1.6.1-bin-hadoop2.6/python/lib/py4j-0.9-src.zip")
sys.path.append("C:/Program Files (x86)/Java/jdk1.8.0_102")

from pyspark import SparkContext
from pyspark import SparkConf

sc = SparkContext("local", "testing")

　　　　在跑算法以前，建议输出Spark Context以下，若是能够正常打印内存地址，则说明Spark的运行环境搞定了。

print sc

　　　　好比个人输出是：

<;pyspark.context.SparkContext object at 0x07352950>;　
　　　　首先咱们将u.data文件读入内存，并尝试输出第一行的数据来检验是否成功读入，注意复制代码的时候，数据的目录要用你本身的u.data的目录。代码以下：

#下面目录要用解压后u.data所在的目录
user_data = sc.textFile("C:/Temp/ml-100k/u.data")
user_data.first()

　　　　输出以下：

u'196\t242\t3\t881250949'
　　　　能够看到数据是用\t分开的，咱们须要将每行的字符串划开，成为数组，并只取前三列，不要时间戳那一列。代码以下：

rates = user_data.map(lambda x: x.split("\t")[0:3])
print rates.first()

　　　　输出以下：

[u'196', u'242', u'3']
　　　　此时虽然咱们已经获得了评分矩阵数组对应的RDD，可是这些数据都仍是字符串，Spark须要的是若干Rating类对应的数组。所以咱们如今将RDD的数据类型作转化，代码以下：

from pyspark.mllib.recommendation import Rating
rates_data = rates.map(lambda x: Rating(int(x[0]),int(x[1]),int(x[2])))
print rates_data.first()

　　　　输出以下：

Rating(user=196, product=242, rating=3.0)
　　　　可见咱们的数据已是基于Rating类的RDD了，如今咱们终于能够把整理好的数据拿来训练了，代码以下, 咱们将矩阵分解的维度设置为20，最大迭代次数设置为5，而正则化系数设置为0.02。在实际应用中，咱们须要经过交叉验证来选择合适的矩阵分解维度与正则化系数。这里咱们因为是实例，就简化了。

from  pyspark.mllib.recommendation import ALS
from pyspark.mllib.recommendation import MatrixFactorizationModel
sc.setCheckpointDir('checkpoint/')
ALS.checkpointInterval = 2
model = ALS.train(ratings=rates_data, rank=20, iterations=5, lambda_=0.02)

　　　　将模型训练完毕后，咱们终于能够来作推荐系统的预测了。

　　　　首先作一个最简单的预测，好比预测用户38对物品20的评分。代码以下：

print model.predict(38,20)

　　　　输出以下：

0.311633491603
　　　　可见评分并不高。

　　　　如今咱们来预测了用户38最喜欢的10个物品，代码以下：

print model.recommendProducts(38,10)

　　　　输出以下：

[Rating(user=38, product=95, rating=4.995227969811873), Rating(user=38, product=304, rating=2.5159673379104484), Rating(user=38, product=1014, rating=2.165428673820349), Rating(user=38, product=322, rating=1.7002266119079879), Rating(user=38, product=111, rating=1.2057528774266673), Rating(user=38, product=196, rating=1.0612630766055788), Rating(user=38, product=23, rating=1.0590775012913558), Rating(user=38, product=327, rating=1.0335651317559753), Rating(user=38, product=98, rating=0.9677333686628911), Rating(user=38, product=181, rating=0.8536682271006641)]

　　　　能够看出用户38可能喜欢的对应评分从高到低的10个物品。

　　　　接着咱们来预测下物品20可能最值得推荐的10个用户，代码以下：

print model.recommendUsers(20,10)

　　　　输出以下：

[Rating(user=115, product=20, rating=2.9892138653406635), Rating(user=25, product=20, rating=1.7558472892444517), Rating(user=7, product=20, rating=1.523935609195585), Rating(user=286, product=20, rating=1.3746309116764184), Rating(user=222, product=20, rating=1.313891405211581), Rating(user=135, product=20, rating=1.254412853860262), Rating(user=186, product=20, rating=1.2194811581542384), Rating(user=72, product=20, rating=1.1651855319930426), Rating(user=241, product=20, rating=1.0863391992741023), Rating(user=160, product=20, rating=1.072353288848142)]

　　　　如今咱们来看看每一个用户最值得推荐的三个物品，代码以下:

print model.recommendProductsForUsers(3).collect()

　　　　因为输出很是长，这里就不将输出copy过来了。

　　　　而每一个物品最值得被推荐的三个用户，代码以下：

print model.recommendUsersForProducts(3).collect()

　　　　一样因为输出很是长，这里就不将输出copy过来了。

　　　　但愿上面的例子对你们使用Spark矩阵分解推荐算法有帮助。

 

（欢迎转载，转载请注明出处。欢迎沟通交流： 微信：nickchen121）