数据下载：
https://github.com/apache/spark/tree/master/data/mllib/als
代码案例参考：
https://github.com/apache/spark/blob/master/examples/src/main/python/ml/als_example.py
需要注意代码里long(p[3])改成float

ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]), rating=float(p[2]), timestamp=float(p[3])))
代码：
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.recommendation import ALS
from pyspark.sql import Row
import os
os.environ["PYSPARK_PYTHON"]="/Users/lonng/opt/anaconda3/python.app/Contents/MacOS/python"


spark = SparkSession\
        .builder\
        .appName("ALSExample")\
        .getOrCreate()

# $example on$
lines = spark.read.text("./sample_movielens_ratings.txt").rdd

parts = lines.map(lambda row: row.value.split("::"))
ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),
                                     rating=float(p[2]), timestamp=float(p[3])))
ratings = spark.createDataFrame(ratingsRDD)
(training, test) = ratings.randomSplit([0.8, 0.2])
als = ALS(maxIter=5, regParam=0.01, userCol="userId", itemCol="movieId", ratingCol="rating",
              coldStartStrategy="drop")
model = als.fit(training)

# Evaluate the model by computing the RMSE on the test data
predictions = model.transform(test)
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                                predictionCol="prediction")
rmse = evaluator.evaluate(predictions)
print("Root-mean-square error = " + str(rmse))

# Generate top 10 movie recommendations for each user
userRecs = model.recommendForAllUsers(10)
userRecs.show()

# Generate top 10 movie recommendations for a specified set of users
users = ratings.select(als.getUserCol()).distinct().limit(3)
userSubsetRecs = model.recommendForUserSubset(users, 10)
userSubsetRecs.show()


# Generate top 10 user recommendations for each movie
movieRecs = model.recommendForAllItems(10)
movieRecs.show()


# Generate top 10 user recommendations for a specified set of movies
movies = ratings.select(als.getItemCol()).distinct().limit(3)
movieSubSetRecs = model.recommendForItemSubset(movies, 10)

movieSubSetRecs.show()




模型评分
# Evaluate the model by computing the RMSE on the test data
predictions = model.transform(test)
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                                predictionCol="prediction")
rmse = evaluator.evaluate(predictions)
print("Root-mean-square error = " + str(rmse))


对于单个user或moive推荐可能的电影或可能的用户,参考http://spark.apache.org/docs/latest/api/python/_modules/pyspark/ml/recommendation.html
￥** 正常第一种方式速度快一些
1、在使用了userRecs = model.recommendForAllUsers(10) ， model.recommendForAllItems(10)后：


userRecs.where(userRecs.userId == 28).select("recommendations.movieId", "recommendations.rating").collect()

movieRecs.where(movieRecs.movieId == 31)\
        .select("recommendations.userId", "recommendations.rating").collect()


2、使用model.recommendForUserSubse、recommendForItemSubsett形式

user_subset = ratings.where(ratings.userId == 28)
model.recommendForUserSubset(user_subset,10).select("recommendations.movieId", "recommendations.rating").first()


item_subset = ratings.where(ratings.item_id == 2)
item_subset_recs = model.recommendForItemSubset(item_subset, 3)
item_subset_recs.select("recommendations.user_id", "recommendations.rating").first()





全部推荐可以下列方式读取，或后续存入数据库比如redis



pandas dataframe保存到redis方式，默认保存到db0，可以StrictRedis改port，db库

模型保存与加载
from pyspark.ml.recommendation import ALS,ALSModel


model_path = temp_path + "/als_model"
model.save(model_path)
model2 = ALSModel.load(model_path)

需要注意代码里long(p[3])改成float

ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]), rating=float(p[2]), timestamp=float(p[3])))代码：

from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.recommendation import ALS
from pyspark.sql import Row
import os
os.environ["PYSPARK_PYTHON"]="/Users/lonng/opt/anaconda3/python.app/Contents/MacOS/python"


spark = SparkSession\
        .builder\
        .appName("ALSExample")\
        .getOrCreate()

# $example on$
lines = spark.read.text("./sample_movielens_ratings.txt").rdd

parts = lines.map(lambda row: row.value.split("::"))
ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),
                                     rating=float(p[2]), timestamp=float(p[3])))
ratings = spark.createDataFrame(ratingsRDD)
(training, test) = ratings.randomSplit([0.8, 0.2])
als = ALS(maxIter=5, regParam=0.01, userCol="userId", itemCol="movieId", ratingCol="rating",
              coldStartStrategy="drop")
model = als.fit(training)

# Evaluate the model by computing the RMSE on the test data
predictions = model.transform(test)
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                                predictionCol="prediction")
rmse = evaluator.evaluate(predictions)
print("Root-mean-square error = " + str(rmse))

# Generate top 10 movie recommendations for each user
userRecs = model.recommendForAllUsers(10)
userRecs.show()

# Generate top 10 movie recommendations for a specified set of users
users = ratings.select(als.getUserCol()).distinct().limit(3)
userSubsetRecs = model.recommendForUserSubset(users, 10)
userSubsetRecs.show()


# Generate top 10 user recommendations for each movie
movieRecs = model.recommendForAllItems(10)
movieRecs.show()


# Generate top 10 user recommendations for a specified set of movies
movies = ratings.select(als.getItemCol()).distinct().limit(3)
movieSubSetRecs = model.recommendForItemSubset(movies, 10)

movieSubSetRecs.show()

模型评分妙笔阁

# Evaluate the model by computing the RMSE on the test data
predictions = model.transform(test)
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                                predictionCol="prediction")
rmse = evaluator.evaluate(predictions)
print("Root-mean-square error = " + str(rmse))

对于单个user或moive推荐可能的电影或可能的用户,参考http://www.mbgtxt.com/docs/latest/api/python/_modules/pyspark/ml/recommendation.html

￥** 正常第一种方式速度快一些

1、在使用了userRecs = model.recommendForAllUsers(10) ， model.recommendForAllItems(10)后：

userRecs.where(userRecs.userId == 28).select("recommendations.movieId", "recommendations.rating").collect()

movieRecs.where(movieRecs.movieId == 31)\
        .select("recommendations.userId", "recommendations.rating").collect()



2、使用model.recommendForUserSubse、recommendForItemSubsett形式

user_subset = ratings.where(ratings.userId == 28)
model.recommendForUserSubset(user_subset,10).select("recommendations.movieId", "recommendations.rating").first()


item_subset = ratings.where(ratings.item_id == 2)
item_subset_recs = model.recommendForItemSubset(item_subset, 3)
item_subset_recs.select("recommendations.user_id", "recommendations.rating").first()

全部推荐可以下列方式读取，或后续存入数据库比如redis

pandas dataframe保存到redis方式，默认保存到db0，可以StrictRedis改port，db库

模型保存与加载

from pyspark.ml.recommendation import ALS,ALSModel


model_path = temp_path + "/als_model"
model.save(model_path)
model2 = ALSModel.load(model_path)

 

1. 协同过滤
“协同过滤”就是协同大家的反馈、评价和意见一起对海量的信息进行过滤，从中筛选出目标用户可能感兴趣的信息的推荐过程。下面介绍基于用户的UserCF：


生成共现矩阵。假设有m个用户，n个物品，每个用户会对n个物品中的一个或者几个进行评分，未评分的物品分值就用问号表示，则所有m个用户对物品的评分可形成一个$m*n$的评分矩阵，也就是协同过滤中的共现矩阵。


生成共现矩阵后，推荐问题就转换成了预测矩阵中问号的值的过程。


通过每个用户对所有物品的评分向量，利用余弦相似度、皮尔逊相关系数（可以引入物品平均分的方式减少物品评分）等，找到与需要推荐用户X最相似的Top n用户


利用用户相似度和相似用户评分的加权平均偶的目标用户的评价预测。下式中，$w_{u,s}$是物品u和用户s的相似度，$u_{s,p}$是用户s对物品p的评分。
$R_{u,p} = \frac{\sum_{s \in S}(w_{u,s} \cdot R_{s,p})}{\sum_{s \in S}w_{u,s}}$


userCF主要有量大缺点，(1) 在互联网场景下，用户数量远大于商品数量，而且还会持续增加，这导致用户相似度矩阵的存储开销很大；(2) 用户的历史数据很稀疏，找到相似用户的准确度很低。
由于上面两大缺点，Amazon最终没有采用该方法，而是采用了ItemCF。ItemCF的计算过程和UserCF类似，但是在应用场景上有所不同。UserCT具备强社交属性，并且更适合发现热点以及跟踪热点的趋势。ItemCF更适用于兴趣变化较为稳定的应用。
协同过滤有两大缺点，(1) 热门的物品具有很强的头部效应，容易跟大量物品产生相似性，而尾部物品则完全相反；(2) 协同过滤仅仅利用用户和物品的交互信息，造成了信息遗漏。
2. 矩阵分解
协同过滤是基于用户的观看历史，找到与目标用户看过同样视频的相似用户，然后找到这些相似用户喜欢看的其他视频，推荐给目标用户。矩阵分解算法则期望为每一位用户和视频生成一个隐向量，将用户和视频定位到隐向量的表示空间上，距离相近的用户和视频表明兴趣特点接近，在推荐过程中，就把距离相近的视频推荐给目标用户。
在矩阵分解的算法框架下，用户和物品的隐向量是通过分解共现矩阵得到的。对矩阵进行矩阵分解的方法主要有三种，特征值分解、奇异值分解、梯度下降。其中，特征值分解只能作用于方阵，故排除。奇异值分解要求原始的共现矩阵是稠密的，所以必须对确实的元素值进行填充，并且计算复杂度为$O(mn^2)$，所以也不合适。因此，梯度下降成了进行矩阵分解的主要方法，目标函数是让原始评分$r_{ui}$与用户向量和物品向量之积$q^T_ip_u$的差尽量小，同时为了减少过拟合现象，加入正则化项。下式中，K是所有用户评分样本的集合。
$\underset{q^*,p^*}{min}\underset{(u,i)\in K}{\sum}(r_{ui}-q_i^Tp_u)^2+\lambda(||q_i||^2+||p_u||^2)$

对$q_i$求偏导，得$\frac{\partial h}{\partial q_i} = 2(r_{ui}-q_i^Tp_u)p_u-2\lambda q_i$
对$p_u$求偏导，得$\frac{\partial h}{\partial q_i} = 2(r_{ui}-q_i^Tp_u)q_i-2\lambda p_u$
沿梯度反方向更新参数

矩阵分解相比协同过滤有三个优点：(1) 泛化能力强，一定程度上解决了数据稀疏问题；(2) 空间复杂度低；(3)更好的扩展性和灵活性

关注我带你拿大厂offer>_<

本文简单扼要地介绍推荐算法中的两种经典算法：协同过滤和矩阵分解。内容有以下三部分
协同过滤算法
矩阵分解
协同过滤与矩阵分解的关系
早期的推荐系统以业务理解为核心，通过复杂的规则描述来向用户推荐商品，这种做法具有透明性高、可移植性差、维护成本高、泛化能力差等特点。在这样一个背景下，协同过滤算法在Amazon推荐系统中的成功落地成为了当时的焦点。
1. 协同过滤算法 (Collaborative Filtering)
[PDF] Amazon.com Recommendations: Item-to-Item Collaborative Filtering | Semantic Scholar​www.semanticscholar.org
“协同过滤“，顾名思义就是协同大家的反馈、评价和一件一起对海量信息进行过滤，进而从中筛选出目标用户可能感兴趣的信息的推荐过程。 —— 王喆《深度学习推荐系统》 
要理解协同过滤算法必须理解其中的两个关键名词：共现矩阵和相似矩阵。
共现矩阵的行坐标为用户、列坐标为物品，矩阵中的值代表某个用户对某个物品的评分值（在不同的业务背景下该评分具有不同的实际意义）。
相似矩阵的定义在UserCF中是用户到用户的相似度矩阵，在ItemCF中则是物品到物品的相似度矩阵。
协同过滤的过程可以大致分为两个阶段：
离线阶段：根据共现矩阵计算出相似度矩阵。
线上阶段：根据相似度矩阵完成TopK推荐问题。
离线阶段
现假设从历史数据中可以获得M个用户对N个物品的正负反馈，将其转化成共现矩阵。
数据：用户到物品的评分矩阵，即一个 MxN 的共现矩阵。
目标：计算出 MxM 的用户相似度矩阵，或者 NxN 的物品相似度矩阵。
UserCF：将共现矩阵视作M个长度为N维的用户向量，根据余弦相似度或皮尔逊相关系数计算出每两个用户之间的相似度，共需计算 
 次，时间复杂度 
ItemCF：将共现矩阵视作N个长度为M维的物品向量，同样根据余弦相似度或皮尔逊相关系数计算出每两个物品之间的相似度，共需计算 
 次，时间复杂度 
线上阶段
数据：一个 MxN 的共现矩阵，MxM 的用户相似度矩阵或者 NxN的物品相似度矩阵。
目标：收到一个用户A（非新用户）的线上请求，为该用户推荐K个用户可能感兴趣的物品。
UserCF：根据用户相似度矩阵，找出与用户A最相似的Top m个用户，然后用下面的伪代码计算出该用户对不同物品的评价预测
# 背景：向userA推荐Top K个物品

# 计算出userA对每个候选item的得分
item_scores = {}
for item in 候选物品集：
    sim_sum = 0
    for sim_user in m个最相似的用户:
        item_scores[item] += sim_user对item的评分 * sim_user与userA的相似度
        sim_sum += sim_user与userA的相似度
    item_scores[item] /= sim_sum
# 找出按score倒排找出Top K个item，代码忽略
ItemCF：假设用户A的正反馈物品集为H，根据物品相似度矩阵，然后用下面的伪代码计算出该用户的Top K
# 背景：向userA推荐Top K个物品

# 计算出userA对每个候选item的得分
item_scores = {}
for item in 候选物品集：
    for item_h in 用户A的正反馈物品集H:
        item_scores[item] += 用户A对item_h的评分 * item与item_h的相似度
# 找出按score倒排找出Top K个item，代码忽略
UserCF与ItemCF的特点
UserCF在“用户对物品的兴趣点比较分散和稀疏”的场景中比ItemCF表现更好，例如新闻推荐。
UserCF需要离线对整个用户群体进行相似度矩阵的计算，从计算成本和存储成本上考虑都不太适合用户数量级远大于Item数量级的场景（如小说推荐），ItemCF会更适合这类场景。
协同过滤的计算方式决定了它推荐出来的物品往往集中在头部那些曝光多的热门商品，即头部效应严重，从长期的角度上并不利于推荐池的多样性。
协同过滤算法在用户、商品达到一定数量级之后，会面临着共现矩阵非常稀疏的问题，而协同过滤在稀疏性较强的数据中难以挖掘到用户真正的潜在兴趣。
为了解决协同过滤存在的泛化性能不强的特点，矩阵分解法被提出。
2. 矩阵分解 (Matrix Factorization)
矩阵分解的含义是在协同过滤算法中的共现矩阵的基础上，加入隐向量的概念，加强了模型处理稀疏矩阵的能力。假设M个用户N个物品，共现矩阵为 MxN 维，那么矩阵分解算法最终会将共现矩阵拆成一个 MxF的用户向量矩阵和一个 NxF的物品向量矩阵（F即隐向量的长度），实际推荐过程中可以直接用线上请求的用户与候选集的物品进行相似度计算，再取Top K。
矩阵分解的过程是怎么样的呢？在具体的算法实现上，矩阵分解主要有三种实现方法。
特征值分解
奇异值分解（SVD）
梯度下降
其中特征值分解只能用于方阵，不适用于大部分场景下的矩阵分解。
奇异值分解的计算方法在这里不详细介绍，它会将一个MxN的矩阵表示成MxM矩阵、MxN对角阵和NxN矩阵的乘积，然后取对角阵的对角线上前F大的元素，使得MxN的矩阵被分解为MxF、FxF和FxN三个矩阵的乘积，也就得到了用户和物品的向量矩阵。奇异值分解的方法本质上要求共现矩阵是稠密的（但大多数互联网场景下用户的行为都是稀疏的），并且时间复杂度为 
 ，在用户量和物品数量很大的情况下计算的时间成本非常高。
梯度下降法是矩阵分解的主流方法，梯度下降法的过程可以概括为以下几步：
确定待优化参数，预测值
     假设 
为用户向量矩阵，
 为物品向量矩阵，那么预测值就是 
 , 
 为用户u对物品i的评分。
2. 确定目标函数
     这里用均方误差，目标函数为 
     为了防止模型在局部参数上过拟合，我们加入正则化项
3. 求偏导，得梯度
     对 
 求偏导，得到的结果为
     对 
 求偏导，得到的结果为
4. 利用梯度的反方向对参数进行更新，从i步到i+1步为
     其中 
 为学习率
3. 协同过滤与矩阵分解的关系
协同过滤算法以用户或物品为单位，找出一个相似的群体，以该群体的整体偏好来估计当前用户的偏好。矩阵分解则借助隐变量，将整个共现矩阵在隐变量的空间内进行描述，用相似度来计算用户对某个商品的兴趣程度。后者在前者的基础上有以下几个点做出了改进：
泛化性能强，解决了数据稀疏的问题。协同过滤算法，以ItemCF为例，是以日志中用户有过正反馈的物品集来刻画用户的兴趣的，如果用户只对极少数物品有过正反馈，那么利用ItemCF的方法来进行推荐对用户兴趣的刻画是不准确的（UserCF中则体现在用户的相似度计算不够准确）。而矩阵分解则以隐变量来刻画用户的兴趣，由于所有用户隐变量的定义都是统一的，所以在计算的过程中，隐变量是可以在所有样本上进行更新的。
拓展性强，中间产物更加灵活。协同过滤算法的中间产物是相似度矩阵，而矩阵分解的产物是用户和物品的向量。相似度矩阵只能用于直接计算推荐结果，但是用户向量和物品向量却还可以进行聚类分析或者作为Embedding特征输入到其他模型当中。
空间复杂度由 
 、 
 降到了 
该文章源自笔者结合自身的理解对相关知识点的整合，从个人的学习笔记的角度落笔，如有错漏不当之处，还望读者批评指正