- 一、离线推荐服务
- 二、离线统计服务
- 1.统计服务主体框架
- 2.基于隐语义模型的协同过滤推荐
- 2.1 用户商品推荐列表
- 2.2 商品相似度矩阵
- 2.3 模型评估与参数选取
离线推荐服务是综合用户所有的历史数据，利用设定的离线统计算法和离线推荐算法周期性的进行结果统计与保存，计算的结果在一定时间周期内是固定不变的，变更的频率取决于算法调度的频率。
离线推荐服务主要计算一些可以预先进行统计和计算的指标，为实时计算和前端业务相应提供数据支撑。
离线推荐服务主要分为统计推荐、基于隐语义模型的协同过滤推荐以及基于内容和基于Item-CF的相似推荐。我们这一章主要介绍前两部分，基于内容和Item-CF的推荐在整体结构和实现上是类似的，我们将在后面详细介绍。
统计服务主体框架包括：
- 历史热门商品统计
- 近期热门商品统计
- 优质商品统计
在recommender下新建子项目StatisticsRecommender，pom.xml文件中只需引入spark、scala和mongodb的相关依赖：
org.apache.spark spark-core_2.11 org.apache.spark spark-sql_2.11 org.scala-lang scala-library org.mongodb casbah-core_2.11 ${casbah.version} org.mongodb.spark mongo-spark-connector_2.11 ${mongodb-spark.version}
在resources文件夹下引入log4j.properties，然后在src/main/scala下新建scala 单例对象com.atguigu.statistics.StatisticsRecommender。
package com.atguigu.statistics
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import java.text.SimpleDateFormat
import java.util.Date
//定义样例类
case class Rating(userId: Int, productId: Int, score: Double, timestamp: Int)
//将mongodb封装成样例类
case class MongoConfig(uri: String, db: String)
object StatisticsRecommender {
//定义要操作的表的名称（读取）
val MONGODB_RATING_COLLECTION = "Rating"
//定义统计的表的名称（写入）
val RATE_MORE_PRODUCTS = "RateMoreProducts" //历史热门商品统计表
val RATE_MORE_RECENTLY_PRODUCTS = "RateMoreRecentlyProducts" //最近热门商品统计表
val AVERAGE_PRODUCTS = "AverageProducts" //商品平均得分统计
def main(args: Array[String]): Unit = {
//配置信息
val config: Map[String, String] = Map(
"spark.cores" -> "local[*]",
"mongo.uri" -> "mongodb://localhost:27017/recommender",
"mongo.db" -> "recommender"
)
//构建sparkContext执行环境入口对象
val conf: SparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("StatisticsRecommender")
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
import spark.implicits._
//声明一个隐式的配置对象,隐式参数--mongodb的连接配置
implicit val mongoConfig: MongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))
//从Mongodb中加载Rating数据,并转化为样例类再转化为dataframe
val ratingDF: DataFrame = spark.read
.option("uri", mongoConfig.uri)
.option("collection", MONGODB_RATING_COLLECTION)
.format("com.mongodb.spark.sql")
.load()
.as[Rating]
.toDF()
//创建一张ratings的临时表
ratingDF.createOrReplaceTempView("ratings")
//TOOD：用spark sql去做不同的统计推荐
//1.历史热门商品，按照评分个数统计,最终得到productId,count
//通过Spark SQL读取评分数据集，统计所有评分中评分数最多的商品，然后按照从大到小排序，将最终结果写入MongoDB的RateMoreProducts数据集中。
val rateMoreProductsDF: DataFrame = spark.sql("select productId,count(productId) as count from ratings group by productId order by count desc")
storeDFInMongoDB(rateMoreProductsDF,RATE_MORE_PRODUCTS)
//2.近期热门商品，把时间戳转换为yyyyMM格式进行评分个数统计
//最终得到productId, count ,yearmonth
//创建一个日期格式化工具
val simpleDataFormat: SimpleDateFormat = new SimpleDateFormat("yyyyMM")
//注册UDF，将timestamp转化为年月格式,将秒数转化为毫秒数再转化为年月格式
spark.udf.register("changeDate", (x:Int) => simpleDataFormat.format(new Date(x * 1000L)).toInt)
// 将原来的Rating数据集中的时间转换成年月的格式
val ratingOfYearMonth = spark.sql("select productId, score, changeDate(timestamp) as yearmonth from ratings")
// 将新的数据集注册成为一张表
ratingOfYearMonth.createOrReplaceTempView("ratingOfMonth")
//按照yearmonth与productId进行聚合，再按照yearmonth与count进行降序排序
val rateMoreRecentlyProducts = spark.sql("select productId, count(productId) as count ,yearmonth from ratingOfMonth group by yearmonth,productId order by yearmonth desc, count desc")
//把df保存到mongodb
storeDFInMongoDB(rateMoreRecentlyProducts,RATE_MORE_RECENTLY_PRODUCTS)
//3.优质商品统计，商品的平均评分，最终得到productId与avg
val avergerProductsDF: DataFrame = spark.sql("select productId,avg(score) as avg from ratings group by productId order by avg desc")
storeDFInMongoDB(avergerProductsDF,AVERAGE_PRODUCTS)
spark.stop()
}
def storeDFInMongoDB(df: DataFrame,collection_name:String)(implicit mongoConfig: MongoConfig): Unit ={
df
.write
.option("uri",mongoConfig.uri)
.option("collection",collection_name)
.mode("overwrite")
.format("com.mongodb.spark.sql")
.save()
}
}
运行结束后，从MongoDB中查看结果：
C:Usersadmin>mongo
MongoDB shell version v4.2.21
connecting to: mongodb://127.0.0.1:27017/?compressors=disabled&gssapiServiceName=mongodb
Implicit session: session { "id" : UUID("68bc0cf1-5482-40e4-b7ab-b9517874077b") }
MongoDB server version: 4.2.21
Server has startup warnings:
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten]
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten] ** WARNING: Access control is not enabled for the database.
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten] ** Read and write access to data and configuration is unrestricted.
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten]
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten] ** WARNING: This server is bound to localhost.
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten] ** Remote systems will be unable to connect to this server.
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten] ** Start the server with --bind_ip to specify which IP
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten] ** addresses it should serve responses from, or with --bind_ip_all to
2022-08-03T00:52:25.812+0800 I CONTROL [initandlisten] ** bind to all interfaces. If this behavior is desired, start the
2022-08-03T00:52:25.813+0800 I CONTROL [initandlisten] ** server with --bind_ip 127.0.0.1 to disable this warning.
2022-08-03T00:52:25.813+0800 I CONTROL [initandlisten]
---
Enable MongoDB's free cloud-based monitoring service, which will then receive and display
metrics about your deployment (disk utilization, CPU, operation statistics, etc).
The monitoring data will be available on a MongoDB website with a unique URL accessible to you
and anyone you share the URL with. MongoDB may use this information to make product
improvements and to suggest MongoDB products and deployment options to you.
To enable free monitoring, run the following command: db.enableFreeMonitoring()
To permanently disable this reminder, run the following command: db.disableFreeMonitoring()
---
> show dbs
admin 0.000GB
config 0.000GB
local 0.000GB
recommender 0.002GB
> use recommender
switched to db recommender
> show tables
AverageProducts
Product
RateMoreProducts
RateMoreRecentlyProducts
Rating
> db.AverageProducts.find().pretty()
{
"_id" : ObjectId("62ea07bacd3bd456c020cb70"),
"productId" : 371783,
"avg" : 4.63035019455253
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb6e"),
"productId" : 473113,
"avg" : 4.641025641025641
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb6a"),
"productId" : 387693,
"avg" : 4.6652892561983474
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb6d"),
"productId" : 459600,
"avg" : 4.810344827586207
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb6f"),
"productId" : 145218,
"avg" : 4.64796633941094
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb72"),
"productId" : 314081,
"avg" : 4.640410958904109
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb74"),
"productId" : 184282,
"avg" : 4.606116774791474
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb71"),
"productId" : 259637,
"avg" : 4.636636636636637
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb6b"),
"productId" : 228884,
"avg" : 4.7330246913580245
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb6c"),
"productId" : 469516,
"avg" : 4.621794871794871
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb73"),
"productId" : 201454,
"avg" : 4.690340909090909
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb69"),
"productId" : 460360,
"avg" : 4.737609329446064
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb75"),
"productId" : 487126,
"avg" : 4.586885245901639
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb77"),
"productId" : 260348,
"avg" : 4.580645161290323
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb76"),
"productId" : 517740,
"avg" : 4.583916083916084
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb79"),
"productId" : 425715,
"avg" : 4.563549160671463
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb7a"),
"productId" : 104168,
"avg" : 4.563342318059299
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb78"),
"productId" : 139060,
"avg" : 4.59214501510574
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb7b"),
"productId" : 508447,
"avg" : 4.554098360655738
}
{
"_id" : ObjectId("62ea07bacd3bd456c020cb7c"),
"productId" : 218439,
"avg" : 4.553429027113237
}
Type "it" for more
> db.RateMoreProducts.find().pretty()
> db.RateMoreRecentlyProducts.find().pretty()
2.基于隐语义模型的协同过滤推荐
项目采用ALS作为协同过滤算法，根据MongoDB中的用户评分表计算离线的用户商品推荐列表以及商品相似度矩阵。
2.1 用户商品推荐列表通过ALS训练出来的Model来计算所有当前用户商品的推荐列表，主要思路如下：
- userId和productId做笛卡尔积，产生（userId，productId）的元组
- 通过模型预测（userId，productId）对应的评分。
- 将预测结果通过预测分值进行排序。
- 返回分值最大的K个商品，作为当前用户的推荐列表。
最后生成的数据结构如下：将数据保存到MongoDB的UserRecs表中。
新建recommender的子项目OfflineRecommender，引入spark、scala、mongo和jblas的依赖：
org.scalanlp jblas ${jblas.version} org.apache.spark spark-core_2.11 org.apache.spark spark-sql_2.11 org.apache.spark spark-mllib_2.11 org.scala-lang scala-library org.mongodb casbah-core_2.11 ${casbah.version} org.mongodb.spark mongo-spark-connector_2.11 ${mongodb-spark.version}
在resources文件夹下引入log4j.properties，然后在src/main/scala下新建scala 单例对象com.atguigu.offline.OfflineRecommender。
package com.atguigu.offline
import org.apache.spark.SparkConf
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.jblas.DoubleMatrix
//定义样例类
case class ProductRating(userId: Int, productId: Int, score: Double, timestamp: Int)
//将mongodb封装成样例类
case class MongoConfig(uri: String, db: String)
//定义标准推荐对象
case class Recommendation(productId: Int, score: Double)
//定义用户推荐列表
case class UserRecs(userId: Int, recs: Seq[Recommendation])
//定义商品相似度列表
case class ProductRecs(productId: Int, recs: Seq[Recommendation])
object OfflineRecommender {
//定义要操作的表的名称（读取）
val MONGODB_RATING_COLLECTION = "Rating"
//定义用户推荐列表
val USER_RECS = "UserRecs"
//定义商品推荐列表
val PRODUCT_RECS = "ProductRecs"
//定义推荐的个数（最大个数）
val USER_MAX_RECOMMENDATION = 20
def main(args: Array[String]): Unit = {
//配置信息
val config: Map[String, String] = Map(
"spark.cores" -> "local[*]",
"mongo.uri" -> "mongodb://localhost:27017/recommender",
"mongo.db" -> "recommender"
)
//构建sparkContext执行环境入口对象
val conf: SparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("OfflineRecommender")
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
import spark.implicits._
//声明一个隐式的配置对象,隐式参数--mongodb的连接配置
implicit val mongoConfig: MongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))
//从Mongodb中加载Rating数据,并转化为样例类再转化为RDD再进行Map转化
//cache()：将RDD保存在内存里面，避免RDD重复计算
val ratingRDD: RDD[(Int, Int, Double)] = spark.read
.option("uri", mongoConfig.uri)
.option("collection", MONGODB_RATING_COLLECTION)
.format("com.mongodb.spark.sql")
.load()
.as[ProductRating]
.rdd
.map(
rating => (rating.userId, rating.productId, rating.score)
).cache()
//提取出所有的productId与UserId
//元组里面取元素_._1
val userRDD: RDD[Int] = ratingRDD.map(_._1).distinct()
val productRDD: RDD[Int] = ratingRDD.map(_._2).distinct()
//TODO:核心计算过程
//1.训练隐语义模型
//将ratingRDD进行转换成ALS模型中需要的数据格式,调用定义好的样例类
val trainDataRDD: RDD[Rating] = ratingRDD.map(x => Rating(x._1, x._2, x._3))
//定义模型训练的参数,rank:表示隐特征的个数；iterations：表示迭代次数；lambda：表示正则化系数
val (rank, iterations, lambda) = (5, 10, 0.01)
val model: MatrixFactorizationModel = ALS.train(trainDataRDD, rank, iterations, lambda)
//2.获得预测评分矩阵，得到用户的推荐列表
//用productRDD与UserRDD做笛卡尔积，得到空的userProductsRDD
val userProducts: RDD[(Int, Int)] = userRDD.cartesian(productRDD)
val preRating: RDD[Rating] = model.predict(userProducts)
//从预测评分矩阵中得到用户推荐列表
//先进行过滤，再转换数据格式，再基于user进行聚合
val userRecs: DataFrame = preRating.filter(_.rating > 0)
.map(
rating => (rating.user, (rating.product, rating.rating))
)
.groupByKey()
.map {
//模式匹配
case (userId, recs) =>
UserRecs(userId, recs.toList.sortWith(_._2 > _._2).take(USER_MAX_RECOMMENDATION).map(x => Recommendation(x._1, x._2)))
}
.toDF()
storeDFInMongoDB(userRecs,USER_RECS)
//3.利用商品的特征向量，计算商品的相似度列表
//调用jblas中相似度计算方法，需要转换成DoubleMatrix数据格式
val productFeatures: RDD[(Int, DoubleMatrix)] = model.productFeatures.map{
case (productId,features) => (productId,new DoubleMatrix(features))
}
//两两配对商品，计算余弦相似度
val productRecs: DataFrame = productFeatures.cartesian(productFeatures)
.filter{
//模式匹配
case (a,b) => a._1 != b._1
}
.map {
case (a,b) =>
val simScore: Double = consinSim(a._2, b._2)
(a._1,(b._1,simScore))
}
.filter(_._2._2>0.6)
.groupByKey()
.map {
//模式匹配
case (productId, recs) =>
ProductRecs(productId, recs.toList.sortWith(_._2 > _._2).map(x => Recommendation(x._1, x._2)))
}
.toDF()
storeDFInMongoDB(productRecs,PRODUCT_RECS)
//关闭spark
spark.stop()
}
//计算余弦相似度
def consinSim(product1: DoubleMatrix,product2: DoubleMatrix): Double = {
product1.dot(product2) / ( product1.norm2() * product2.norm2() )
}
def storeDFInMongoDB(df: DataFrame,collection_name:String)(implicit mongoConfig: MongoConfig): Unit ={
df
.write
.option("uri",mongoConfig.uri)
.option("collection",collection_name)
.mode("overwrite")
.format("com.mongodb.spark.sql")
.save()
}
}
通过查看MongoDB中的UserRecs表，其结果为：
> show tables
AverageProducts
Product
ProductRecs
RateMoreProducts
RateMoreRecentlyProducts
Rating
UserRecs
> db.UserRecs.find().pretty()
{
"_id" : ObjectId("62ea8b6acd3bd43a60b17c51"),
"userId" : 37025,
"recs" : [
{
"productId" : 406697,
"score" : 7.4438937113164005
},
{
"productId" : 156103,
"score" : 5.843661037687788
},
{
"productId" : 469516,
"score" : 5.181569233058891
},
{
"productId" : 231449,
"score" : 5.122126051481281
},
{
"productId" : 13543,
"score" : 5.032388591578158
},
{
"productId" : 410608,
"score" : 4.929721517995789
},
{
"productId" : 139060,
"score" : 4.850941805388885
},
{
"productId" : 8195,
"score" : 4.845108431240849
},
{
"productId" : 21643,
"score" : 4.779079014119098
},
{
"productId" : 109236,
"score" : 4.740873200632537
},
{
"productId" : 57272,
"score" : 4.740065595781363
},
{
"productId" : 473113,
"score" : 4.725606778937141
},
{
"productId" : 116405,
"score" : 4.724732418069501
},
{
"productId" : 300265,
"score" : 4.712243032015264
},
{
"productId" : 520943,
"score" : 4.50634726581384
},
{
"productId" : 508447,
"score" : 4.421818154794035
},
{
"productId" : 442353,
"score" : 4.3343509369932605
},
{
"productId" : 183679,
"score" : 4.326064248055729
},
{
"productId" : 410716,
"score" : 4.322165047600581
},
{
"productId" : 507644,
"score" : 4.131260228388946
}
]
}
{
"_id" : ObjectId("62ea8b6acd3bd43a60b17c52"),
"userId" : 41765,
"recs" : [
{
"productId" : 154318,
"score" : 6.444461446808766
},
{
"productId" : 365357,
"score" : 5.172166568523835
},
{
"productId" : 119166,
"score" : 5.06825060131694
},
{
"productId" : 425715,
"score" : 5.065625275066604
},
{
"productId" : 457976,
"score" : 5.042510774724991
},
{
"productId" : 367460,
"score" : 4.954262313663024
},
{
"productId" : 352021,
"score" : 4.914302174473304
},
{
"productId" : 21643,
"score" : 4.889703511278231
},
{
"productId" : 155001,
"score" : 4.88767650078044
},
{
"productId" : 14103,
"score" : 4.863698023520641
},
{
"productId" : 62138,
"score" : 4.798401136870837
},
{
"productId" : 406697,
"score" : 4.774858717517169
},
{
"productId" : 33864,
"score" : 4.639764344470233
},
{
"productId" : 379243,
"score" : 4.445774920342778
},
{
"productId" : 487126,
"score" : 4.425731156112397
},
{
"productId" : 6797,
"score" : 4.385734287996334
},
{
"productId" : 228884,
"score" : 4.276207579442106
},
{
"productId" : 459600,
"score" : 4.256293363940154
},
{
"productId" : 13316,
"score" : 4.242235580513345
},
{
"productId" : 387693,
"score" : 4.234428198410992
}
]
}
2.2 商品相似度矩阵
通过ALS计算商品相似度矩阵，该矩阵用于查询当前商品的相似商品并为实时推荐系统服务。
离线计算的ALS 算法，算法最终会为用户、商品分别生成最终的特征矩阵，分别是表示用户特征矩阵的U(m x k)矩阵，每个用户由 k个特征描述；表示物品特征矩阵的V(n x k)矩阵，每个物品也由 k 个特征描述。
V(n x k)表示物品特征矩阵，每一行是一个k 维向量，虽然我们并不知道每一个维度的特征意义是什么，但是k 个维度的数学向量表示了该行对应商品的特征。所以，每个商品用V(n x k)每一行的向量
<
t
1
,
t
2
,
.
.
.
,
t
k
>
数据集中任意两个商品间相似度都可以由公式计算得到，商品与商品之间的相似度在一段时间内基本是固定值。最后生成的数据保存到MongoDB的ProductRecs表中。
其核心代码为：
//3.利用商品的特征向量，计算商品的相似度列表
//调用jblas中相似度计算方法，需要转换成DoubleMatrix数据格式
val productFeatures: RDD[(Int, DoubleMatrix)] = model.productFeatures.map{
case (productId,features) => (productId,new DoubleMatrix(features))
}
//两两配对商品，计算余弦相似度
val productRecs: DataFrame = productFeatures.cartesian(productFeatures)
.filter{
//模式匹配
case (a,b) => a._1 != b._1
}
.map {
case (a,b) =>
val simScore: Double = consinSim(a._2, b._2)
(a._1,(b._1,simScore))
}
.filter(_._2._2>0.6)
.groupByKey()
.map {
//模式匹配
case (productId, recs) =>
ProductRecs(productId, recs.toList.sortWith(_._2 > _._2).map(x => Recommendation(x._1, x._2)))
}
.toDF()
通过查看MongoDB中的ProductRecs表，其结果为：
> show tables
AverageProducts
Product
ProductRecs
RateMoreProducts
RateMoreRecentlyProducts
Rating
UserRecs
> db.ProductRecs.find().pretty()
{
"_id" : ObjectId("62ea9006cd3bd408382b039f"),
"productId" : 474201,
"recs" : [
{
"productId" : 203971,
"score" : 0.9999581062826495
},
{
"productId" : 3982,
"score" : 0.9687890562617127
},
{
"productId" : 75701,
"score" : 0.9420241333154399
},
{
"productId" : 154318,
"score" : 0.9405515635348338
},
{
"productId" : 443787,
"score" : 0.889002891492535
},
{
"productId" : 300265,
"score" : 0.8835959834195055
},
{
"productId" : 206404,
"score" : 0.8692129871878008
},
{
"productId" : 145218,
"score" : 0.8665773218739673
},
{
"productId" : 260348,
"score" : 0.8514561529176196
},
{
"productId" : 314081,
"score" : 0.8470445393915133
},
{
"productId" : 371783,
"score" : 0.817488508475942
},
{
"productId" : 460360,
"score" : 0.8096484290490837
},
{
"productId" : 48907,
"score" : 0.7991868382197964
},
{
"productId" : 469516,
"score" : 0.7855925651666726
},
{
"productId" : 407816,
"score" : 0.7741929819234256
},
{
"productId" : 459600,
"score" : 0.7714808773145543
},
{
"productId" : 13543,
"score" : 0.7697317660546396
},
{
"productId" : 122417,
"score" : 0.7596865452035727
},
{
"productId" : 228884,
"score" : 0.7441174782020797
},
{
"productId" : 507644,
"score" : 0.7317715827209548
},
{
"productId" : 492579,
"score" : 0.7312896609252877
},
{
"productId" : 473113,
"score" : 0.7213895096508246
},
{
"productId" : 160597,
"score" : 0.7210934512772513
},
{
"productId" : 487126,
"score" : 0.7090094722019329
},
{
"productId" : 489369,
"score" : 0.7037617363913163
},
{
"productId" : 499898,
"score" : 0.6962167820997771
},
{
"productId" : 407523,
"score" : 0.6911425389166844
},
{
"productId" : 275707,
"score" : 0.6764143954626965
},
{
"productId" : 253869,
"score" : 0.6746947973237932
},
{
"productId" : 104168,
"score" : 0.6575175992736674
},
{
"productId" : 520943,
"score" : 0.6517642879852018
},
{
"productId" : 387693,
"score" : 0.6441234816258334
},
{
"productId" : 383502,
"score" : 0.6387597297660523
},
{
"productId" : 204025,
"score" : 0.6358141933271599
},
{
"productId" : 410608,
"score" : 0.6284766944417334
},
{
"productId" : 425715,
"score" : 0.6243394899690763
},
{
"productId" : 323519,
"score" : 0.6232032699031145
},
{
"productId" : 457976,
"score" : 0.6160485917546723
},
{
"productId" : 259637,
"score" : 0.6011662714834244
}
]
}
2.3 模型评估与参数选取
在上述模型训练的过程中，我们直接给定了隐语义模型的rank、iterations、lambda三个参数。对于我们的模型，这并不一定是最优的参数选取，所以我们需要对模型进行评估。通常的做法是计算均方根误差（RMSE），考察预测评分与实际评分之间的误差。
有了RMSE，我们可以就可以通过多次调整参数值，来选取RMSE最小的一组作为我们模型的优化选择。
在scala/com.atguigu.offline/下新建单例对象ALSTrainer，代码主体架构如下：
package com.atguigu.offline
import breeze.numerics.sqrt
import com.atguigu.offline.OfflineRecommender.MONGODB_RATING_COLLECTION
import org.apache.spark.SparkConf
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
object ALSTrainer {
def main(args: Array[String]): Unit = {
//配置信息
val config: Map[String, String] = Map(
"spark.cores" -> "local[*]",
"mongo.uri" -> "mongodb://localhost:27017/recommender",
"mongo.db" -> "recommender"
)
//构建SparkSession执行环境入口对象
val conf: SparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("OfflineRecommender")
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
import spark.implicits._
//声明一个隐式的配置对象,隐式参数--mongodb的连接配置
implicit val mongoConfig: MongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))
//从Mongodb中加载Rating数据,并直接转化为ALS需要的Rating数据
//cache()：将RDD保存在内存里面，避免RDD重复计算
val ratingRDD: RDD[Rating] = spark.read
.option("uri", mongoConfig.uri)
.option("collection", MONGODB_RATING_COLLECTION)
.format("com.mongodb.spark.sql")
.load()
.as[ProductRating]
.rdd
.map(
rating => Rating(rating.userId, rating.productId, rating.score)
).cache()
// 将一个RDD随机切分成两个RDD，用以划分训练集和测试集
val splits: Array[RDD[Rating]] = ratingRDD.randomSplit(Array(0.8, 0.2))
val trainingRDD: RDD[Rating] = splits(0)
val testingRDD: RDD[Rating] = splits(1)
//输出最优参数
//adjustALSParams方法是模型评估的核心，输入一组训练数据和测试数据，输出计算得到最小RMSE的那组参数
adjustALSParams(trainingRDD, testingRDD)
//关闭Spark
spark.close()
}
def adjustALSParams(trainingRDD: RDD[Rating], testingRDD: RDD[Rating]): Unit ={
//遍历数组中定义的参数取值，并用yield关键字。存储每一次循环的中间结果
// 这里指定迭代次数为5，rank和lambda在几个值中选取调整
val result = for(rank <- Array(5,10,20,50); lambda <- Array(1, 0.1, 0.01, 0.001))
yield {
val model: MatrixFactorizationModel = ALS.train(trainingRDD,rank,5,lambda)
val rmse: Double = getRMSE(model, testingRDD)
(rank,lambda,rmse)
}
// 输出最优参数
println(result.minBy(_._3))
}
def getRMSE(model: MatrixFactorizationModel, data: RDD[Rating]): Double ={
//构建userProducts，得到预测评分矩阵
val userProducts: RDD[(Int, Int)] = data.map(item => (item.user,item.product))
val predictRating: RDD[Rating] = model.predict(userProducts)
//得到实际评分
val real: RDD[((Int, Int), Double)] = data.map(item => ((item.user,item.product),item.rating))
//得到预测评分
val predict: RDD[((Int, Int), Double)] = predictRating.map(item => ((item.user,item.product),item.rating))
// 按照公式计算RMSE，首先将预测评分与实际评分按照(userId,productId)做一个连接
sqrt(
real.join(predict).map{
case ((userId,productId),(real,pre))=>
// 真实值和预测值之间的差
val err: Double = real - pre
err * err
}.mean()
)
}
}
运行代码，我们就可以得到目前数据的最优模型参数。
(5,0.1,1.3376743677938532)
