协同过滤 - 基于 RDD 的 API
协同过滤
协同过滤 通常用于推荐系统。这些技术旨在填充用户-项目关联矩阵中缺失的条目。spark.mllib 目前支持基于模型的协同过滤，其中用户和产品由一小组潜在因子描述，这些因子可用于预测缺失的条目。spark.mllib 使用 交替最小二乘法 (ALS) 算法来学习这些潜在因子。spark.mllib 中的实现具有以下参数
- numBlocks 是用于并行计算的块数（设置为 -1 以自动配置）。
- rank 是要使用的特征数（也称为潜在因子数）。
- iterations 是要运行的 ALS 迭代次数。ALS 通常在 20 次迭代或更少迭代内收敛到合理的解。
- lambda 指定 ALS 中的正则化参数。
- implicitPrefs 指定是使用_显式反馈_ ALS 变体还是适用于_隐式反馈_数据的变体。
- alpha 是适用于 ALS 的隐式反馈变体的参数，它控制着对偏好观察的_基线_置信度。
显式反馈与隐式反馈
基于矩阵分解的协同过滤的标准方法将用户-项目矩阵中的条目视为用户对项目给出的_显式_偏好，例如，用户对电影的评分。
在许多现实世界的用例中，通常只能访问_隐式反馈_（例如，浏览、点击、购买、点赞、分享等）。spark.mllib 中用于处理此类数据的方法取自 隐式反馈数据集的协同过滤。本质上，这种方法不是尝试直接对评分矩阵建模，而是将数据视为表示用户行为观察_强度_的数字（例如点击次数，或观看电影的累计持续时间）。然后，这些数字与观察到的用户偏好的置信度相关联，而不是与对项目给出的明确评分相关联。然后，模型尝试找到可用于预测用户对项目的预期偏好的潜在因子。
正则化参数的缩放
从 v1.1 开始，我们在解决每个最小二乘问题时，通过用户在更新用户因子时生成的评分数或产品在更新产品因子时收到的评分数来缩放正则化参数 lambda。这种方法被称为“ALS-WR”，并在论文“Netflix 奖的大规模并行协同过滤”中进行了讨论。它使 lambda 不太依赖于数据集的规模，因此我们可以将从采样子集中学习到的最佳参数应用于完整数据集，并期望获得类似的性能。
示例
在以下示例中，我们加载评分数据。每行包含一个用户、一个产品和一个评分。我们使用默认的 ALS.train() 方法，该方法假设评分是显式的。我们通过测量评分预测的均方误差来评估推荐。
有关 API 的更多详细信息，请参阅 ALS Python 文档。
from pyspark.mllib.recommendation import ALS, MatrixFactorizationModel, Rating
# Load and parse the data
data = sc.textFile("data/mllib/als/test.data")
ratings = data.map(lambda l: l.split(','))\
.map(lambda l: Rating(int(l[0]), int(l[1]), float(l[2])))
# Build the recommendation model using Alternating Least Squares
rank = 10
numIterations = 10
model = ALS.train(ratings, rank, numIterations)
# Evaluate the model on training data
testdata = ratings.map(lambda p: (p[0], p[1]))
predictions = model.predictAll(testdata).map(lambda r: ((r[0], r[1]), r[2]))
ratesAndPreds = ratings.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)
MSE = ratesAndPreds.map(lambda r: (r[1][0] - r[1][1])**2).mean()
print("Mean Squared Error = " + str(MSE))
# Save and load model
model.save(sc, "target/tmp/myCollaborativeFilter")
sameModel = MatrixFactorizationModel.load(sc, "target/tmp/myCollaborativeFilter")如果评分矩阵是从其他信息源派生的（即，它是从其他信号推断出来的），则可以使用 trainImplicit 方法获得更好的结果。
# Build the recommendation model using Alternating Least Squares based on implicit ratings
model = ALS.trainImplicit(ratings, rank, numIterations, alpha=0.01)在以下示例中，我们加载评分数据。每行包含一个用户、一个产品和一个评分。我们使用默认的 ALS.train() 方法，该方法假设评分是显式的。我们通过测量评分预测的均方误差来评估推荐模型。
有关 API 的更多详细信息，请参阅 ALS Scala 文档。
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
import org.apache.spark.mllib.recommendation.Rating
// Load and parse the data
val data = sc.textFile("data/mllib/als/test.data")
val ratings = data.map(_.split(',') match { case Array(user, item, rate) =>
Rating(user.toInt, item.toInt, rate.toDouble)
})
// Build the recommendation model using ALS
val rank = 10
val numIterations = 10
val model = ALS.train(ratings, rank, numIterations, 0.01)
// Evaluate the model on rating data
val usersProducts = ratings.map { case Rating(user, product, rate) =>
(user, product)
}
val predictions =
model.predict(usersProducts).map { case Rating(user, product, rate) =>
((user, product), rate)
}
val ratesAndPreds = ratings.map { case Rating(user, product, rate) =>
((user, product), rate)
}.join(predictions)
val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =>
val err = (r1 - r2)
err * err
}.mean()
println(s"Mean Squared Error = $MSE")
// Save and load model
model.save(sc, "target/tmp/myCollaborativeFilter")
val sameModel = MatrixFactorizationModel.load(sc, "target/tmp/myCollaborativeFilter")如果评分矩阵是从另一个信息源派生的（即，它是从其他信号推断出来的），则可以使用 trainImplicit 方法获得更好的结果。
val alpha = 0.01
val lambda = 0.01
val model = ALS.trainImplicit(ratings, rank, numIterations, lambda, alpha)所有 MLlib 的方法都使用 Java 友好类型，因此您可以在 Java 中以与 Scala 相同的方式导入和调用它们。唯一需要注意的是，这些方法采用 Scala RDD 对象，而 Spark Java API 使用单独的 JavaRDD 类。您可以通过在 JavaRDD 对象上调用 .rdd() 将 Java RDD 转换为 Scala RDD。下面给出了一个与 Scala 中提供的示例等效的自包含应用程序示例
有关 API 的更多详细信息，请参阅 ALS Java 文档。
import scala.Tuple2;
import org.apache.spark.api.java.*;
import org.apache.spark.mllib.recommendation.ALS;
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
import org.apache.spark.mllib.recommendation.Rating;
import org.apache.spark.SparkConf;
SparkConf conf = new SparkConf().setAppName("Java Collaborative Filtering Example");
JavaSparkContext jsc = new JavaSparkContext(conf);
// Load and parse the data
String path = "data/mllib/als/test.data";
JavaRDD<String> data = jsc.textFile(path);
JavaRDD<Rating> ratings = data.map(s -> {
String[] sarray = s.split(",");
return new Rating(Integer.parseInt(sarray[0]),
Integer.parseInt(sarray[1]),
Double.parseDouble(sarray[2]));
});
// Build the recommendation model using ALS
int rank = 10;
int numIterations = 10;
MatrixFactorizationModel model = ALS.train(JavaRDD.toRDD(ratings), rank, numIterations, 0.01);
// Evaluate the model on rating data
JavaRDD<Tuple2<Object, Object>> userProducts =
ratings.map(r -> new Tuple2<>(r.user(), r.product()));
JavaPairRDD<Tuple2<Integer, Integer>, Double> predictions = JavaPairRDD.fromJavaRDD(
model.predict(JavaRDD.toRDD(userProducts)).toJavaRDD()
.map(r -> new Tuple2<>(new Tuple2<>(r.user(), r.product()), r.rating()))
);
JavaRDD<Tuple2<Double, Double>> ratesAndPreds = JavaPairRDD.fromJavaRDD(
ratings.map(r -> new Tuple2<>(new Tuple2<>(r.user(), r.product()), r.rating())))
.join(predictions).values();
double MSE = ratesAndPreds.mapToDouble(pair -> {
double err = pair._1() - pair._2();
return err * err;
}).mean();
System.out.println("Mean Squared Error = " + MSE);
// Save and load model
model.save(jsc.sc(), "target/tmp/myCollaborativeFilter");
MatrixFactorizationModel sameModel = MatrixFactorizationModel.load(jsc.sc(),
"target/tmp/myCollaborativeFilter");要运行上述应用程序，请按照 Spark 快速入门指南的 自包含应用程序 部分中提供的说明进行操作。请确保还将_spark-mllib_作为依赖项包含在您的构建文件中。
教程
2014 年 Spark 峰会的培训练习 包括一个关于 使用 spark.mllib 进行个性化电影推荐 的动手教程。