用 Spark ALS 训练并评估电影推荐

本文完整翻译并整理 Apache Spark 官方 Collaborative Filtering，覆盖算法参数、显式与隐式反馈、正则化、冷启动及 Python、Scala、Java、R 四组示例。原页未列个人作者，发布维护方为 Apache Software Foundation。2026-10-09 重新读取 latest 页面并核对全文，页面标题标示 Spark 4.2.0；可用 4.2.0 固定版本页面复核。复现时应锁定版本，不能让 latest 链接随升级改变环境。

协同过滤如何工作

协同过滤常用于推荐系统，目标是补全用户与物品关联矩阵中的缺失项。spark.ml 支持基于模型的协同过滤：以少量潜在因子表示用户与物品，用这些因子预测未观察到的关联。Spark 采用交替最小二乘法（ALS）学习潜在因子。它交替固定一侧因子、求解另一侧的最小二乘问题，逐轮改进矩阵分解。

DataFrame 版 ALS 的用户和物品 ID 必须是整数意义上的标识，值处于有符号32位整数范围内。列可以使用其他数值类型，但不能因此放入超出整数范围的 ID；字符串 ID 应先建立稳定映射。上表默认值勘误来源为 Spark 4.2.0 ALS Python API。

显式反馈与隐式反馈

标准矩阵分解把用户—物品矩阵中的值视为明确表达的偏好，例如用户给电影打的分数。实际业务常常只有浏览、点击、购买、点赞或分享等隐式反馈。Spark 对隐式反馈的处理来自论文 Collaborative Filtering for Implicit Feedback Datasets。

这种方法不直接把点击次数或观看时长当作评分，而把它们理解为行为观测的强度，由此表示对偏好判断的置信程度。模型寻找能预测用户对物品预期偏好的潜在因子。隐式反馈的建模目标与评分预测不同，不能把一次评分 RMSE 的改善直接等同于推荐排序质量提高。

正则化参数的缩放

每次求解最小二乘问题时，Spark 会按该用户给出的评分数量缩放正则化项；更新物品因子时，则按该物品收到的评分数量缩放。这一方法称为 ALS-WR，见 Large-Scale Parallel Collaborative Filtering for the Netflix Prize。

这种缩放减小了 regParam 对数据规模的依赖，因此可以把样本子集上选出的参数应用于完整数据，期望获得相近表现。这是算法设计动机，不是对任意数据分布或生产效果的保证，仍需重新验证。

冷启动策略

预测时，测试数据可能包含训练中未出现的用户或物品：生产环境中新用户、新物品没有评分历史；交叉验证中的随机划分也很容易把某个用户或物品的记录全部分到评估集。缺少对应因子时，ALSModel.transform 默认产生 NaN。线上系统可以借此识别冷启动并选择回退策略。

但 RegressionEvaluator 等评估器遇到 NaN 后，评分指标也可能变成 NaN，无法据此选择模型。将 coldStartStrategy 设为 drop，可以删除预测为 NaN 的行，使指标只在有预测值的记录上计算。目前指南列出的策略是 nan（默认）与 drop。

drop 不会解决冷启动。它只是缩小了参与评估的数据集合，因此必须同时报告测试记录总数、成功评分数和覆盖率。否则低 RMSE 可能只是因为困难样本被排除了。真实系统仍需要热门物品、内容特征或其他回退方案。

Python：评分输入、训练、评估和推荐

样例读取 Spark 随附的评分文件，每行按 userId::movieId::rating::timestamp 排列。原指南将其作为 MovieLens 评分示例；完整 MovieLens 数据的来源和使用条件应另查 GroupLens，不能把 Spark 的代码许可证等同于所有第三方数据的许可证。

下方保留官方算法与数据流程，补齐配套 Python 文件中的 SparkSession、结果展示与 stop；注释译为中文，移除只用于文档抽取的标记和 main 包装。数据路径相对当前工作目录，单独安装 PySpark 不代表该样例文件已存在。默认 implicitPrefs=False，把输入当作显式评分；测试集用评分预测的均方根误差（RMSE）评估。

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

spark = SparkSession.builder.appName("ALSExample").getOrCreate()
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=int(p[3])))
ratings = spark.createDataFrame(ratingsRDD)
(training, test) = ratings.randomSplit([0.8, 0.2])

# 用训练数据建立 ALS 模型；drop 只删除无法预测的行
als = ALS(maxIter=5, regParam=0.01, userCol="userId",
          itemCol="movieId", ratingCol="rating",
          coldStartStrategy="drop")
model = als.fit(training)

# 对保留下来的测试记录计算评分 RMSE
predictions = model.transform(test)
evaluator = RegressionEvaluator(
    metricName="rmse", labelCol="rating", predictionCol="prediction")
rmse = evaluator.evaluate(predictions)
print("Root-mean-square error = " + str(rmse))

# 每个用户的十部候选电影，以及每部电影的十名候选用户
userRecs = model.recommendForAllUsers(10)
movieRecs = model.recommendForAllItems(10)
# 指定用户、指定电影的子集
users = ratings.select(als.getUserCol()).distinct().limit(3)
userSubsetRecs = model.recommendForUserSubset(users, 10)
movies = ratings.select(als.getItemCol()).distinct().limit(3)
movieSubSetRecs = model.recommendForItemSubset(movies, 10)
userRecs.show()
movieRecs.show()
userSubsetRecs.show()
movieSubSetRecs.show()
spark.stop()

完整源文件：examples/src/main/python/ml/als_example.py。这里的 recommendForAllUsers 为每个已训练用户产生电影候选；recommendForAllItems 为每部已训练电影产生用户候选；两个 Subset 方法则只处理给定集合。distinct().limit(3) 没有排序，不能把这三个 ID 当作稳定固定的测试名单。候选结果还需要过滤已消费物品，并应用业务可用性和权限约束。

如果数据由其他行为信号推导而来，可以改用隐式反馈模式。原文表述为可能改善效果；仍须按实际目标验证，不能保证一定更好：

als = ALS(maxIter=5, regParam=0.01, implicitPrefs=True,
          userCol="userId", itemCol="movieId", ratingCol="rating")

这一单独片段没有设置 drop，仍使用默认 nan；评估时要明确选择策略。切换 implicitPrefs 后，rating 列的业务含义也应改为相应的行为强度，而不能仅切换一个开关就认定输入已适配。

Scala 示例

这组代码完成相同流程：读取四字段评分、划分训练集与测试集、训练默认显式反馈 ALS、计算 RMSE，再生成全部与子集的双向推荐。片段假定已有 SparkSession spark 和相应隐式转换；完整文件为 ALSExample.scala。

import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.ALS

case class Rating(userId: Int, movieId: Int, rating: Float, timestamp: Long)
def parseRating(str: String): Rating = {
  val fields = str.split("::")
  assert(fields.size == 4)
  Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat, fields(3).toLong)
}
val ratings = spark.read.textFile("data/mllib/als/sample_movielens_ratings.txt")
  .map(parseRating).toDF()
val Array(training, test) = ratings.randomSplit(Array(0.8, 0.2))
val als = new ALS()
  .setMaxIter(5).setRegParam(0.01)
  .setUserCol("userId").setItemCol("movieId").setRatingCol("rating")
val model = als.fit(training)
// 删除冷启动产生的 NaN 后，再计算 RMSE
model.setColdStartStrategy("drop")
val predictions = model.transform(test)
val evaluator = new RegressionEvaluator()
  .setMetricName("rmse").setLabelCol("rating").setPredictionCol("prediction")
val rmse = evaluator.evaluate(predictions)
println(s"Root-mean-square error = $rmse")
val userRecs = model.recommendForAllUsers(10)
val movieRecs = model.recommendForAllItems(10)
val users = ratings.select(als.getUserCol).distinct().limit(3)
val userSubsetRecs = model.recommendForUserSubset(users, 10)
val movies = ratings.select(als.getItemCol).distinct().limit(3)
val movieSubSetRecs = model.recommendForItemSubset(movies, 10)

隐式反馈版本只需在构造模型时显式指定：

val als = new ALS()
  .setMaxIter(5).setRegParam(0.01).setImplicitPrefs(true)
  .setUserCol("userId").setItemCol("movieId").setRatingCol("rating")

Java 示例

Java 版本用可序列化的 Rating 类与 getter 建立 DataFrame。以下保留原页示例结构，补充 Dataset 和 Row 导入；Rating 是外围示例类的静态嵌套类，训练语句放在方法体中，它不是可以直接粘贴为单个顶层类的完整文件。完整应用见 JavaALSExample.java。

import java.io.Serializable;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.recommendation.ALS;
import org.apache.spark.ml.recommendation.ALSModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public static class Rating implements Serializable {
  private int userId;
  private int movieId;
  private float rating;
  private long timestamp;
  public Rating() {}
  public Rating(int userId, int movieId, float rating, long timestamp) {
    this.userId = userId;
    this.movieId = movieId;
    this.rating = rating;
    this.timestamp = timestamp;
  }
  public int getUserId() { return userId; }
  public int getMovieId() { return movieId; }
  public float getRating() { return rating; }
  public long getTimestamp() { return timestamp; }
  public static Rating parseRating(String str) {
    String[] fields = str.split("::");
    if (fields.length != 4) {
      throw new IllegalArgumentException("Each line must contain 4 fields");
    }
    int userId = Integer.parseInt(fields[0]);
    int movieId = Integer.parseInt(fields[1]);
    float rating = Float.parseFloat(fields[2]);
    long timestamp = Long.parseLong(fields[3]);
    return new Rating(userId, movieId, rating, timestamp);
  }
}
// 以下语句放在已创建 SparkSession spark 的方法体中
JavaRDD<Rating> ratingsRDD = spark.read()
  .textFile("data/mllib/als/sample_movielens_ratings.txt").javaRDD()
  .map(Rating::parseRating);
Dataset<Row> ratings = spark.createDataFrame(ratingsRDD, Rating.class);
Dataset<Row>[] splits = ratings.randomSplit(new double[]{0.8, 0.2});
Dataset<Row> training = splits[0];
Dataset<Row> test = splits[1];
ALS als = new ALS().setMaxIter(5).setRegParam(0.01)
  .setUserCol("userId").setItemCol("movieId").setRatingCol("rating");
ALSModel model = als.fit(training);
model.setColdStartStrategy("drop");
Dataset<Row> predictions = model.transform(test);
RegressionEvaluator evaluator = new RegressionEvaluator()
  .setMetricName("rmse").setLabelCol("rating").setPredictionCol("prediction");
double rmse = evaluator.evaluate(predictions);
System.out.println("Root-mean-square error = " + rmse);
Dataset<Row> userRecs = model.recommendForAllUsers(10);
Dataset<Row> movieRecs = model.recommendForAllItems(10);
Dataset<Row> users = ratings.select(als.getUserCol()).distinct().limit(3);
Dataset<Row> userSubsetRecs = model.recommendForUserSubset(users, 10);
Dataset<Row> movies = ratings.select(als.getItemCol()).distinct().limit(3);
Dataset<Row> movieSubSetRecs = model.recommendForItemSubset(movies, 10);

隐式反馈的构造方式如下，冷启动与评分覆盖率的注意事项不变：

ALS als = new ALS()
  .setMaxIter(5).setRegParam(0.01).setImplicitPrefs(true)
  .setUserCol("userId").setItemCol("movieId").setRatingCol("rating");

R 示例

原页还给出 SparkR 的最小示例，API 详见 spark.als，配套文件为 als.R。它假定 SparkR 会话已经初始化。

# 载入训练数据
data <- list(list(0, 0, 4.0), list(0, 1, 2.0), list(1, 1, 3.0),
             list(1, 2, 4.0), list(2, 1, 1.0), list(2, 2, 5.0))
df <- createDataFrame(data, c("userId", "movieId", "rating"))
training <- df
test <- df
# 拟合、查看模型摘要并预测
model <- spark.als(training, maxIter = 5, regParam = 0.01,
                   userCol = "userId", itemCol = "movieId", ratingCol = "rating")
summary(model)
predictions <- predict(model, test)
head(predictions)

这段 R 代码把同一个 DataFrame 同时赋给 training 与 test，因此只是演示拟合和预测 API，不构成独立测试集评估。不能把它的预测表现当作泛化指标。

编者补充：报告覆盖率，并说明评估的范围

以下补充代码只做静态审核，未执行。它放在 Python 示例中模型建立之后、会话关闭之前，报告 drop 后还剩多少测试行；没有成功预测时不调用 RMSE。与原例相比新增计数和缓存，并在使用后释放缓存。

# 编者补充：放在 Python 示例 model 建立后、spark.stop() 之前
# 原始 ratings 输入应稳定；固定种子只帮助复现，并不修复所有非确定性输入
test_count = test.count()
predictions = model.transform(test).cache()
scored_count = predictions.count()
coverage = scored_count / test_count if test_count else None
print("test_rows =", test_count, "scored_rows =", scored_count,
      "coverage =", coverage)
if scored_count:
    print("RMSE =", evaluator.evaluate(predictions))
else:
    print("没有可评估记录，不能报告 RMSE")
predictions.unpersist()

原例 randomSplit 未固定种子；如需重复比较，记录数据版本、划分策略与种子，并在模型中明确 seed。官方 API 还要求 ALS 输入 DataFrame 是确定性的：例如重新分区后再进行依赖行顺序的抽样可能导致拟合失败，应按场景固定或检查点化输入。固定种子本身不保证全部处理链确定。

输入解析器直接转 int/float，遇到缺字段、非数字、空值、NaN、无穷大或超范围 ID 会失败或带来无效数据。生产输入应先核对字段数、取值、重复评分和异常值；这里保留示例的简洁解析，不伪装成生产清洗流程。样例不包含网络调用、密钥或动态执行输入字符串，未发现命令注入路径。

这份小数据示例与随机划分只能验证工作流。真实推荐应考虑时间划分、用户与物品覆盖、离线排序指标以及线上目标；已消费物品的过滤应放在候选输出流程中。本文没有运行 Spark，没有给出任何虚构的 RMSE、耗时、推荐列表或性能保证。

来源、引用和许可

Apache Spark，Copyright 2014 and onwards The Apache Software Foundation。本文保留 Apache License 2.0 许可信息；Apache License 2.0 全文见LICENSE-2.0.txt；Spark v4.2.0 NOTICE 全文及来源署名文件分别见本站NOTICE.txt、ATTRIBUTION.txt。MovieLens 等第三方数据的使用条件须另行核验，不能以 Spark 代码许可替代。本稿为未完纪中文翻译整理，主要改动是补齐 Python 外围代码、中文注释、Java 导入、默认值勘误、覆盖率补充和原创图。原文引用的 ALS、隐式反馈和 ALS-WR 论文链接均保留。

Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.