一. 需求

近期朋友问我spark的推荐算法相关的。

二. 解决方案

因为之前没有接触过推荐算法相关，所以我在spark的官网上找了下，结果找到一个非常nice的案例。
https://spark.apache.org/docs/2.4.7/ml-collaborative-filtering.html

2.1 Spark官网demo

2.1.1 协同过滤

协同过滤通常用于推荐系统。这些技术旨在填补用户-项目关联矩阵中缺失的条目。Spark.ML 目前支持基于模型的协同过滤，其中用户和产品由一组潜在因素描述，可用于预测缺失条目 Spark.ML 使用交替最小二乘(ALS)算法来学习这些潜在因子。spark中的实现。Ml有以下参数:

1. numBlocks是将用户和条目划分为并行计算的块的数量(默认为10)。

2. Rank是模型中潜在因子的数量(默认为10)。

3. maxIter是要运行的最大迭代次数(默认为10)。

4. regParam指定ALS中的正则化参数(默认为1.0)。

5. implicitPrefs指定是使用显式反馈ALS变体还是适合隐式反馈数据的变体(默认为false，这意味着使用显式反馈)。

6. alpha是一个适用于ALS隐式反馈变体的参数，它控制着偏好观察的基线置信度(默认为1.0)。

7. Nonnegative指定是否对最小二乘使用非负约束(默认为false)。

注意:
ALS的基于数据框架的API目前只支持用户id和条目id的整数。用户和项目id列也支持其他数字类型，但是id必须在整数值范围内。

2.1.2 显性和隐性反馈

基于矩阵分解的协同过滤的标准方法将用户-物品矩阵中的条目视为用户给物品的显式偏好，例如，用户给电影打分。

在许多现实世界的用例中，只有隐式反馈(如浏览、点击、购买、喜欢、分享等)是很常见的。spark中使用的方法。用于处理此类数据的ml取自隐式反馈数据集协同过滤。从本质上来说，这种方法并不是直接模拟评级矩阵，而是将数据视为代表用户行为观察强度的数字(如点击次数，或某人观看电影的累计时长)。然后，这些数字与观察到的用户偏好的信心水平相关，而不是明确的给商品评级。然后，该模型试图找到潜在的因素，可以用来预测用户对某项商品的预期偏好。

2.1.3 正则化参数的缩放

在解决每个最小二乘问题时，我们将正则化参数regParam缩放为用户在更新用户因子中生成的评级数量，或产品在更新产品因子中收到的评级数量。这种方法被命名为“ALS-WR”，并在《Netflix奖的大规模并行协同过滤》一文中进行了讨论。它减少了regParam对数据集规模的依赖，因此我们可以将从采样子集学到的最佳参数应用到整个数据集，并期望类似的性能。

2.1.4 本身的策略

当使用ALSModel进行预测时，通常会遇到测试数据集中在训练模型期间不存在的用户和/或项。这通常发生在两种情况下:

1. 在生产中，对于新用户或没有评级历史且模型没有经过训练的项目(这就是“冷启动问题”)。

2. 在交叉验证期间，数据在训练集和评估集之间分割。在Spark的CrossValidator或TrainValidationSplit中使用简单的随机分割时，经常会遇到评估集中的用户和/或项目不在训练集中的情况

默认情况下，Spark在ALSModel期间分配NaN预测。当模型中不存在用户和/或项因素时进行转换。这在生产系统中可能很有用，因为它指示了一个新用户或项，因此系统可以决定使用某些回退作为预测。

然而，这在交叉验证期间是不希望的，因为任何NaN预测值都会导致评估度量的NaN结果(例如使用RegressionEvaluator时)。这使得模型选择变得不可能。

Spark允许用户将coldStartStrategy参数设置为“drop”，以便删除DataFrame中任何包含NaN值的预测行。评估指标将在非nan数据上计算，并且是有效的。下面的示例说明了该参数的用法。

注意:
目前支持的冷启动策略是" nan "(上面提到的默认行为)和" drop "。未来可能会支持进一步的策略。

2.1.5 Python代码

在下面的示例中，我们从MovieLens数据集中加载评级数据，每行包含一个用户、一个电影、一个评级和一个时间戳。然后我们训练一个ALS模型，该模型默认情况下假设评分是显式的(implicitPrefs为False)。我们通过测量评级预测的均方根误差来评估推荐模型。

from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.recommendation import ALS
from pyspark.sql import Row

lines = spark.read.text("data/mllib/als/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=long(p[3])))
ratings = spark.createDataFrame(ratingsRDD)
(training, test) = ratings.randomSplit([0.8, 0.2])

# Build the recommendation model using ALS on the training data
# Note we set cold start strategy to 'drop' to ensure we don't get NaN evaluation metrics
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)
# Generate top 10 user recommendations for each movie
movieRecs = model.recommendForAllItems(10)

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

2.2 ALS算法简要解释

ALS是交替最小二乘（alternating least squares）的简称。在机器学习的范畴中，ALS特指使用交替最小二乘求解的一个协同推荐算法。它通过观察到的所有用户给产品的打分，来推断每个用户的喜好并向用户推荐适合的产品。不过ALS无法准确评估新加入的用户或商品。这个问题也被称为Cold Start问题。

2.2.1 举例

ALS推荐算法是基于矩形分解的一种方法。先看看矩阵分解的含义。

我们拿电影推荐作为例子。推荐所使用的数据可以抽象成一个[m,n]的矩阵R，R的每一行代表m个用户对所有电影的评分，n列代表每部电影对应的得分。R是个稀疏矩阵，一个用户只是对所有电影中的一小部分看过，有评分。通过矩阵分解方法，我可以把这个低秩的矩阵，分解成两个小矩阵的点乘。公式如下：

image.png

矩阵R（评分为1、2、3、4、5）：

image.png

我把矩阵分解之后，就变成了下面两个小矩阵（F是隐藏特征）：

image.png

用户的特征

image.png

电影的特征：

1. 分解之后的矩阵，变成了根据特征数决定维数的向量。这种求出的向量作为用户的特征，用在推荐上，被成为隐藏特征或者隐藏因子。

2. 为什么进行矩阵分解呢？因为推荐使用的矩阵不仅是稀疏的而且往往是低秩的。矩阵分解相当于进行了特征提取或者数据的降维。

3. 为了求出两个分解后的矩阵，我可以产生两个维度一样的随机矩阵U和V，点乘之后得到同样m行n列的矩阵R1. 这一步我已经得到两个[m,n]的矩阵，其中一个是反映用户的真实喜好的数据，矩阵R。另一份只是一个近似数据，矩阵R1。我可以找到一个公式来衡量，两个同阶的矩阵的相似程度：

   
   image.png

这是一个损失函数，我的目的就是让这个函数的值最小化，使得我构造的矩阵能够最接近原始矩阵。

2.2.2 ALS算法参数

关键代码(Java)：

MatrixFactorizationModel model = ALS.train(ratings, rank, iterations,lambda);

解释：

1. ratings：训练集，数据格式：（用户id 物品id 评分 ）

2. rank：矩阵分解时对应的低维的维数，即特征向量维数或者说特征数。如果这个值太小拟合的就会不够，误差就很大；如果这个值很大，就会导致模型大泛化能力较差。这个值会影响矩阵分解的性能，越大则算法运行的时间和占用的内存可能会越多。通常需要进行调参，一般可以取10-200之间的数。

3. iterations：在矩阵分解用交替最小二乘法求解时，进行迭代的最大次数。这个值取决于评分矩阵的维度，以及评分矩阵的系数程度。一般来说，不需要太大，比如5-20次即可。

4. lambda：正则因子。lambda也是和rank一样的，如果设置很大就可以防止过拟合问题，如果设置很小，其实可以理解为直接设置为0，那么就不会有防止过拟合的功能了；怎么设置呢？可以从0.0001 ，0.0003，0.001，0.003，0.01，0.03，0.1，0.3，1，3，10这样每次大概3倍的设置，先大概看下哪个值效果比较好，然后在那个比较好的值（比如说0.01）前后再设置一个范围，比如（0.003，0.3）之间，间隔设置小点，即0.003，0.005，0.007，0.009，0.011等等等。

调优:
需要引入均方根误差（RMSE）：均方根误差是用来衡量观测值同真值之间的偏差。

用途：
预测评分和推荐物品或者用户等。

1） 预测用户对物品的评分

predict(int user, int product)

2）预测用户集对物品集的评分

predict(JavaPairRDD<Integer, Integer> usersProducts)

3）推荐用户k个物品

recommendProducts(final int user, int num)

4）对物品推荐k个用户

recommendUsers(final int product, int num)

5） 对所有用户推荐物品，物品数量取前k个

recommendProductsForUsers(int num)

6） 对所有物品推荐用户，用户数量取前k个

recommendUsersForProducts(int num)

参考:

1. https://spark.apache.org/docs/2.4.7/ml-collaborative-filtering.html
2. https://www.jianshu.com/p/cfcbf08900dd