开源精选 - Spark个性化推荐算法

源码：https://github.com/wolf-song-ml/RecommendationSystem

实战篇

1 项目技术架构

2 项目涉及关键技术

Redis：存储用户最近评测队列
Mongdb：BI可视化查询
Elastic Search：文本关键词模糊检索索引、类别完全匹配检索、More like this基于内容推荐api
Flume：实时评测数据采集
Kafka：采集数据中间消息通道 Kafka stream：消息转发中间管道
Spark：spark sql、spark stream、spark M数据统计、加载数据源引擎、机器学习模型
ScalaNLP：JAVA矩阵计算

理论篇

1 推荐系统的意义 - 解决信息过载

搜索引擎时代

分类导航：雅虎
搜索：谷歌、百度

个性化时代(提高用户粘度、增加营收)

系统自动推荐相关的东西：今日头条、豆瓣、电商

2 推荐系统的分类

基于人口统计学的推荐
基于内容的推荐
基于协同过滤的推荐

3 基于人口统计学的推荐

基于人口统计学的推荐机制（Demographic-based Recommendation）是一种最易于实现的推荐方法，它只是简单的根据系统用户的基本信息发现用户的相关程度，然后将相似用户喜爱的其他物品推荐给当前用户。

4 基于内容的推荐

4.1 定义

基于内容的推荐是在推荐引擎出现之初应用最为广泛的推荐机制，它的核心思想是根据推荐物品或内容的元数据，发现物品或者内容的相关性，然后基于用户以往的喜好记录，推荐给用户相似的物品。

4.2 算法流程

对于物品的特征提取——打标签（tag）
对于文本信息的特征提取——关键词
生成分词特征向量矩阵
计算相似度，常用余弦相似度

4.3 核心代码

4.3.1 spark TF-IDF

// 核心部分： 用TF-IDF从内容信息中提取电影特征向量
val movieRecs = movieFeatures.cartesian(movieFeatures).filter{// 把自己跟自己的配对过滤掉case (a, b) => a._1 != b._1}.map{case (a, b) => {val simScore = this.consinSim(a._2, b._2)( a._1, ( b._1, simScore ) )}}.filter(_._2._2 > 0.6)    // 过滤出相似度大于0.6的.groupByKey().map{case (mid, items) => MovieRecs( mid, items.toList.sortWith(_._2 > _._2).map(x => Recommendation(x._1, x._2)) )}.toDF()

4.3.2 ElasticSearch More like this

MoreLikeThisQueryBuilder query = QueryBuilders.moreLikeThisQuery(/*new String[]{"name", "descri", "genres", "actors", "directors", "tags"},*/new MoreLikeThisQueryBuilder.Item[]{new MoreLikeThisQueryBuilder.Item(Constant.ES_INDEX,Constant.ES_MOVIE_TYPE, String.valueOf(mid))});

5 基于协同过滤的推荐

5.1基于用户的协同过滤(UserCF)

计算用户的相似度，推荐相似用户的喜好

5.2 基于物品的协同过滤(ItemCF重点)

计算物品的相似度，推荐相似度高的物品(不同于基于内容的推荐)

5.2.1核心算法：计算同现相似度

5.2.2 核心算法实例

// ( mid1, (mid2, score) )
val simDF = cooccurrenceDF.map{row =>val coocSim = cooccurrenceSim( row.getAs[Long]("cocount"), row.getAs[Long]("count1"),row.getAs[Long]("count2") )( row.getInt(0), ( row.getInt(1), coocSim ) )
}.rdd.groupByKey().map{case (mid, recs) =>MoviesRecs( mid, recs.toList.sortWith(_._2>_._2).take(MAX_RECOMMENDATION).map(x=>Recommendation(x._1,x._2)) )}.toDF()

5.3 基于隐语义算法模型推荐

5.3.1 思想

找到隐藏因子，可以对user和item进行关联

5.3.2 算法公式

5.3.3 核心算法实例

// 训练隐语义模型：Rating(user:Int, product:Int, rating:Double)
val trainData = ratingRDD.map(x => Rating(x._1, x._2, x._3))
// 多个变量赋值
val (rank, iterations, lambda) = (200, 5, 0.1)
val model = ALS.train(trainData, rank, iterations, lambda)// 从rating数据中提取所有的uid和mid，并去重
val userRDD = ratingRDD.map(_._1).distinct()
val movieRDD = ratingRDD.map(_._2).distinct()
val userMovies = userRDD.cartesian(movieRDD)// 调用model的predict方法预测评分
val preRatings = model.predict(userMovies)val userRecs = preRatings.filter(_.rating > 0).map(rating => (rating.user, (rating.product, rating.rating))) // Rating->(uid, (mid, score)).groupByKey().map {case (uid, recs) => UserRecs(uid, recs.toList.sortWith(_._2 > _._2).take(USER_MAX_RECOMMENDATION).map(x => Recommendation(x._1, x._2)))}.toDF()

5.3.4 LFM模型评估、迭代与调优

通常的做法是计算均方根误差（RMSE），考察预测评分与实际评分之间的误差。

有了RMSE，我们可以就可以通过多次调整参数值，来选取RMSE最小的一组作为我们模型的优化选择。

 /*** LFM迭代调参* @param trainData* @param testData*/def adjustALSParam(trainData: RDD[Rating], testData: RDD[Rating]): Unit = {val result = for (rank <- Array(50, 100, 200, 300); lambda <- Array(0.01, 0.1, 1))yield {val model = ALS.train(trainData, rank, 5, lambda)val rmse = getRMSE(model, testData)(rank, lambda, rmse)}// 控制台打印输出最优参数println(result.minBy(_._3))}/*** 均方误差的根* @param model* @param data* @return*/def getRMSE(model: MatrixFactorizationModel, data: RDD[Rating]): Double = {// 计算预测评分val userProducts = data.map(item => (item.user, item.product))val predictRating = model.predict(userProducts)// 以uid，mid作为外键，inner join实际观测值和预测值val actual = data.map(item => ((item.user, item.product), item.rating))val predict = predictRating.map(item => ((item.user, item.product), item.rating))// 内连接得到(uid, mid),(actual, predict)sqrt(actual.join(predict).map {case ((uid, mid), (actual, pre)) => val err = actual - pre; err * err}.mean())}

6 实时推荐算法

6.1算法设计

首先，获取用户u 按时间顺序最近的K 个评分，记为RK；获取商品p 的最相似的K 个商品集合，记为S；
然后，对于每个商品qS ，计算其推荐优先级，计算公式如下：

其中：
表示用户u 对商品r 的评分；
sim(q,r)表示商品q 与商品r 的相似度，设定最小相似度为0.6，当商品q和商品r 相似度低于0.6 的阈值，则视为两者不相关并忽略；
sim_sum 表示q 与RK 中商品相似度大于最小阈值的个数；
incount 表示RK 中与商品q 相似的、且本身评分较高（>=3）的商品个数；
recount 表示RK 中与商品q 相似的、且本身评分较低（❤️）的商品个数；

6.2 削弱因子

将增强因子增加到上述的预测评分中，并减去削弱因子，得到最终的q 商品对于u 的推荐优先级。在计算完每个候选商品q 的后，将生成一组<商品q 的ID, q 的推荐优先级>的列表updatedList：

 /*** 计算备选元素与用户最近评分物品相似度+加强减弱因子:核心算法* @param candidateMovies* @param userRecentlyRatings* @param simMovies* @return*/def computeMovieScores(candidateMovies: Array[Int], userRecentlyRatings: Array[(Int, Double)],simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Array[(Int, Double)] = {val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()// 增强减弱因子val increMap = scala.collection.mutable.HashMap[Int, Int]()val decreMap = scala.collection.mutable.HashMap[Int, Int]()for (candidateMovie <- candidateMovies; userRecentlyRating <- userRecentlyRatings) {val simScore = getMoviesSimScore(candidateMovie, userRecentlyRating._1, simMovies)if (simScore > 0.7) {scores += ((candidateMovie, simScore * userRecentlyRating._2))if (userRecentlyRating._2 > 3) {increMap(candidateMovie) = increMap.getOrElse(candidateMovie, 0) + 1} else {decreMap(candidateMovie) = decreMap.getOrElse(candidateMovie, 0) + 1}}}scores.groupBy(_._1).map {case (mid, scoreList) =>(mid, scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrElse(mid, 1)) - log(decreMap.getOrElse(mid, 1)))}.toArray.sortWith(_._2 > _._2)}

7 冷启动问题

整个推荐系统更多的是依赖于用于的偏好信息进行商品的推荐，那么就会存在一个问题，对于新注册的用户是没有任何偏好信息记录的，那这个时候推荐就会出现问题，导致没有任何推荐的项目出现。
处理这个问题一般是通过当用户首次登陆时，为用户提供交互式的窗口来获取用户对于物品的偏好，让用户勾选预设的兴趣标签。当获取用户的偏好之后，就可以直接给出相应类型商品的推荐