第8章 Spark MLlib机器学习算法库

Spark MLlib实现了ALS算法，它是机器学习的协同过滤算法。机器学习的协同过滤式推荐算法是通过观察所有用户给产品的评分来推断每个用户的喜好，并向用户推荐合适的产品。

在电影推荐系统中，通常分为针对用户推荐电影和针对电影推荐用户两种方式。若采用基于用户的推荐模型，则会利用相似用户的评级来计算对某个用户的推荐。若采用基于物品的推荐模型，则会依靠用户接触过的物品与候选物品之间的相似度来获得推荐。

MLlib实现电影推荐

获取数据点击此处

1. 编写程序，训练模型

编写程序训练模型的代码

$ spark-shell --master local[2]

# 采用Spark-Shell读取u.data数据文件且将其转换为RDD
scala> val dataRdd = sc.textFile("/spark/mldata/ml-100k/u.data")
scala> dataRdd.first()
res0: String = 196      242     3       881250949

# 从上一步得知，该数据由用户id、电影id、等级评价和时间戳组成，不需要用到时间戳。使用take()方法提取前三个字段
# 使用\t进行分割返回一个Array{String}类型RDD，用dataRdds进行接收
scala> val dataRdds = dataRdd.map(_.split("\t").take(3))
scala> dataRdds.first()
res1: Array[String] = Array(196, 242, 3)

# 导入MLlib实现的ALS算法模型库和Rating包
scala> import org.apache.spark.mllib.recommendation.ALS
scala> import org.apache.spark.mllib.recommendation.Rating

# 将dataRdds转换成Rating格式的数据
scala> val ratings = dataRdds.map { case Array(user,movie,rating) =>
                                                            Rating(user.toInt,movie.toInt,rating.toDouble)}
scala> ratings.first()
res6: org.apache.spark.mllib.recommendation.Rating = Rating(196,242,3.0)

# 使用train()函数训练模型
scala> val model = ALS.train(ratings,50,10,0.01)
model: org.apache.spark.mllib.recommendation.MatrixFactorizationModel =
org.apache.spark.mllib.recommendation.MatrixFactorizationModel@6580f76c

2. 为用户推荐多个电影

为用户推荐多个电影的代码

# 针对id为100的用户，预测对id为207的电影喜欢，推荐评级为5.70分
# 定义用户id和推荐数量，调用函数model.recommendProducts()进行实现
scala> val userid = 100
scala> val num = 1
scala> val topRecoPro = model.recommendProducts(userid,num)
topRecoPro: Array[org.apache.spark.mllib.recommendation.Rating] = Array(Rating(100,207,5.704436943409341))

# 为了更直观的检测推荐效果，将u.item文件中的电影id与电影名称进行映射。
# 下述代码的作用是：针对id为100的用户，预测对name为Cyrano de Bergerac (1990)的电影喜欢，推荐评级为5.70分

# 先读取u.item文件并转换为RDD
scala> val moviesRdd =sc.textFile("/spark/mldata/ml-100k/u.item")
# 使用map()函数针对每一项数据进行转换
val titles = moviesRdd.map(line => line.split("\\|").take(2)).map(array=>
                                           (array(0).toInt,array(1))).collectAsMap()
# 通过Rating对象的rating属性来对推荐的电影名称进行匹配
scala> topRecoPro.map(rating => (titles(rating.product),rating.rating))
                                                               .foreach(println)
(Cyrano de Bergerac (1990),5.704436943409341)

3. 将物品推荐给用户

将电影推荐给多个用户的代码

# 对id为100这个电影感兴趣的2名用户如下：
scala> model.recommendUsers(100,2)
res1: Array[org.apache.spark.mllib.recommendation.Rating] = Array(
Rating(495,100,6.541442448267074), Rating(30,100,6.538178750321883))