协同过滤是一种基于一组兴趣相同的用户或项目进行的推荐，它根据邻居用户(与目标用户兴趣相似的用户)的偏好信息产生对目标用户的推荐列表。
frompyspark.ml.evaluationimportRegressionEvaluatorfrompyspark.ml.recommendationimportALSfrompyspark.sqlimportRow根据数据结构创建读取规范创建一个函数，返回即[Int,Int,Float,Long]的对象
deff(x):rel={}rel['userId']=int(x[0])rel['movieId']=int(x[1])rel['rating']=float(x[2])rel['timestamp']=float(x[3])returnrel读取数据ratings=sc.textFile("file:///usr/local/spark/data/mllib/als/sample_movielens_ratings.txt").map(lambdaline:line.split('::')).map(lambdap:Row(**f(p))).toDF()然后把数据打印出来：
ratings.show()+-------+------+-------------+------+|movieId|rating|timestamp|userId|+-------+------+-------------+------+|2|3.0|1.424380312E9|0||3|1.0|1.424380312E9|0||5|2.0|1.424380312E9|0||9|4.0|1.424380312E9|0||11|1.0|1.424380312E9|0||12|2.0|1.424380312E9|0||15|1.0|1.424380312E9|0||17|1.0|1.424380312E9|0||19|1.0|1.424380312E9|0||21|1.0|1.424380312E9|0||23|1.0|1.424380312E9|0||26|3.0|1.424380312E9|0||27|1.0|1.424380312E9|0||28|1.0|1.424380312E9|0||29|1.0|1.424380312E9|0||30|1.0|1.424380312E9|0||31|1.0|1.424380312E9|0||34|1.0|1.424380312E9|0||37|1.0|1.424380312E9|0||41|2.0|1.424380312E9|0|+-------+------+-------------+------+onlyshowingtop20rows构建模型把MovieLens数据集划分训练集和测试集
training,test=ratings.randomSplit([0.8,0.2])使用ALS来建立推荐模型，这里我们构建了两个模型，一个是显性反馈，一个是隐性反馈
alsExplicit=ALS(maxIter=5,regParam=0.01,userCol="userId",itemCol="movieId",ratingCol="rating")alsImplicit=ALS(maxIter=5,regParam=0.01,implicitPrefs=True,userCol="userId",itemCol="movieId",ratingCol="rating")在ML中的实现有如下的参数:
numBlocks是用于并行化计算的用户和商品的分块个数(默认为10)。rank是模型中隐语义因子的个数（默认为10）。maxIter是迭代的次数（默认为10）。regParam是ALS的正则化参数（默认为1.0）。implicitPrefs决定了是用显性反馈ALS的版本还是用适用隐性反馈数据集的版本（默认是false，即用显性反馈）。alpha是一个针对于隐性反馈ALS版本的参数，这个参数决定了偏好行为强度的基准（默认为1.0）。nonnegative决定是否对最小二乘法使用非负的限制（默认为false）。可以调整这些参数，不断优化结果，使均方差变小。比如：imaxIter越大，regParam越小，均方差会越小，推荐结果较优。
接下来，把推荐模型放在训练数据上训练：
modelExplicit=alsExplicit.fit(training)modelImplicit=alsImplicit.fit(training)模型预测使用训练好的推荐模型对测试集中的用户商品进行预测评分，得到预测评分的数据集
predictionsExplicit=modelExplicit.transform(test)predictionsImplicit=modelImplicit.transform(test)我们把结果输出，对比一下真实结果与预测结果：
evaluator=RegressionEvaluator().setMetricName("rmse").setLabelCol("rating").setPredictionCol("prediction")