目录

原因

由于实时性，所以算法设计需要满足一下两点

算法设计

算法实现

算法公式

 完整代码

---

原因

用户对电影的偏好随着时间的推移总是会发生变化的。此时离线系统无法解决，需要实时推荐。

由于实时性，所以算法设计需要满足一下两点

1 根据用户最近几次评分进行推荐；

2 计算量要小，满足响应时间上面的实时；

算法设计

借鉴基于物品Item-CF的算法，将相似度和评分结合起来。

算法实现

1 当用户完成一次评分后，从Redis里取最近的k次评分的数据(mid,score);

2 备选电影(与评分电影的相似的电影)和评分过的电影，根据电影形似度矩阵计算相似度；

3 为提高精准率和召回率，设置偏移项，进行模型的鼓励与惩罚。

算法公式

其中：

表示用户 u 对电影 r 的评分；

sim(q,r)表示电影 q 与电影 r 的相似度，设定最小相似度为 0.6，当电影 q 和 电影 r 相似度低于 0.6 的阈值，则视为两者不相关并忽略；

sim_sum 表示 q 与 RK 中电影相似度大于最小阈值的个数；

incount 表示 RK 中与电影 q 相似的、且本身评分较高（>=3）的电影个数；

recount 表示 RK 中与电影 q 相似的、且本身评分较低（<3）的电影个数；

核心代码

 def computeMovieScore(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.getOrDefault(candidateMovie, 0) + 1
        } else {
          decreMap(candidateMovie) = decreMap.getOrDefault(candidateMovie, 0) + 1
        }
      }
    }
    scores.groupBy(_._1).map{
      case (mid, scoreList) =>
        (mid,scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrDefault(mid, 1)) - log(decreMap.getOrDefault(mid, 1)) )
    }.toArray
  }

 完整代码

package com.qh.streaming

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**
 * 连接助手
 * 序列化
 */
object ConnHelper extends Serializable {
  lazy val jedis = new Jedis("hadoop100") //redis
  lazy val mongoClient = MongoClient(MongoClientURI("mongodb://hadoop100:27017/recommender"))
}

case class MongoConfig(uri: String, db: String)

// 定义一个基准推荐对象
case class Recommendation(mid: Int, score: Double)

// 定义基于预测评分的用户推荐列表
case class UserRecs(uid: Int, recs: Seq[Recommendation])

// 定义基于LFM电影特征向量的电影相似度列表
case class MovieRecs(mid: Int, recs: Seq[Recommendation])

object StreamingRecommender {

  val MAX_USER_RATINGS_NUM = 20
  val MAX_SIM_MOVIES_NUM = 20
  val MONGODB_STREAM_RECS_COLLECTION = "StreamRecs"
  val MONGODB_RATING_COLLECTION = "Rating"
  val MONGODB_MOVIE_RECS_COLLECTION = "MovieRecs"

  def main(args: Array[String]): Unit = {
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://hadoop100:27017/recommender",
      "mongo.db" -> "recommender",
      "kafka.topic" -> "recommender"
    )

    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("StreamingRecommender")
    //    spark2.x sparksession已封装除了 sparkstream上下文之外的 session
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // 拿到streaming context
    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc, Seconds(2)) // batch duration 批处理时间 微批次

    import spark.implicits._

    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    //加载电影相似度矩阵，广播
    val simMovieMatrix = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_MOVIE_RECS_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[MovieRecs]
      .rdd //为了查询相似度的过程更快，转换成map
      .map {
        movieRecs =>
        (movieRecs.mid, movieRecs.recs.map(x => (x.mid, x.score)).toMap)
      }.collectAsMap()

    val simMovieMatrixBroadCast = sc.broadcast(simMovieMatrix)

    // 定义kafka连接参数
    val kafkaParam = Map(
      "bootstrap.servers" -> "hadoop100:9092",
      "key.deserializer" -> classOf[StringDeserializer],  //反序列化
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "recommender",
      "auto.offset.reset" -> "latest"   //偏移量初始化设置
    )
    //kafka创建一个DStream
    val kafkaStream = KafkaUtils.createDirectStream[String, String](
      ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](Array(config("kafka.topic")), kafkaParam )
    )

    //把原始数据 UID|MID|SCORE|TIMESTAMP =>评分流
    val ratingStream = kafkaStream.map {
      msg =>
        val attr = msg.value().split("\\|")
        (attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt)
    }

    //继续做流式处理，核心实时算法
    ratingStream.foreachRDD{
      rdds => rdds.foreach{
        case (uid,mid, score, timestamp) =>
          println("rating data coming ! >>>>>>>>>>>>>>>>")

          //1 从redis里获取当前用户最近的k次评分，保存成Array[(mid, score)]
          val userRecentlyRatings = getUserRecentlyRating( MAX_USER_RATINGS_NUM, uid, ConnHelper.jedis )
          //2 从相似度矩阵中获取当前电影最相似的N个电影，作为备选列表，Array[mid]
//              根据uid过滤，将该用户评分过的电影过滤掉
//              在mongodb里面进行过滤
          val candidateMovies = getTopSimMovies( MAX_SIM_MOVIES_NUM, mid, uid, simMovieMatrixBroadCast.value )
          //3 最每个备选电影计算推荐优先级，得到当前用户的实时推荐列表，Array[(mid, score)]
          val streamResc = computeMovieScore( candidateMovies, userRecentlyRatings, simMovieMatrixBroadCast.value )
          //4 把推荐数据保存到mongodb
          saveDataToMongoDB( uid, streamResc )

      }
    }

    ssc.start()



    println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> streaming started")
    ssc.awaitTermination()



  }
//  redis返回的是java类，需要引入转换类
  import scala.collection.JavaConversions._

  /**
   * 从 redis 中选取 该用户最近 num次的 对某个电影的评分
   * @param num  选取的最近评分记录的 数量
   * @param uid  该用户的 ID
   * @param jedis
   * @return 该用户的最近num次 的评分记录
   */
  def getUserRecentlyRating(num: Int, uid: Int, jedis: Jedis): Array[(Int, Double)] = {
    //从redis读数据，用户评分数据保存在 uid:UID 为Key的队列里， value MID:SCORE
    jedis.lrange("uid:" + uid, 0, num-1)
      .map{
        item =>
          val attr = item.split("\\:")
          ( attr(0).trim.toInt, attr(1).trim.toDouble )  //uid ， score
      }
      .toArray
  }

  /**
   * 根据当前电影 从相似度矩阵 选取除了用户已看的电影 之外的 num个形似的电影列表
   * @param num   要选取相似的电影数量
   * @param mid   当前低钠盐的ID
   * @param uid   当前评分的用户ID
   * @param simMovies  相似度矩阵
   * @param mongoConfig
   * @return  过滤后的 备选电影的列表
   */
  def getTopSimMovies(num: Int, mid: Int, uid: Int, simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                     (implicit mongoConfig: MongoConfig): Array[Int] = {
    //从相似度矩阵中取出所有相似的电影
    val allSimMovies: Array[(Int, Double)] = simMovies(mid).toArray
    //从mongodb中查询用户已看过的电影, 用于过滤
    val ratingExist = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
      .find( MongoDBObject("uid" -> uid) )
      .toArray
      .map{
        item => item.get("mid").toString.toInt
      }
    //过于该用户看过的电影(用户如果评过分数 就代表用户看过该电影)
    allSimMovies.filter( x => ! ratingExist.contains(x._1) )
      .sortWith(_._2>_._2)
      .take(num)
      .map( x => x._1 )
  }



  def computeMovieScore(candidateMovies: Array[Int],
                        userRecentlyRatings: Array[(Int, Double)],
                        simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Array[(Int, Double)] = {
    //定义一个ArrayBuffer.用户保存每一个备选电影的基础得分
    val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()
    // 定义一个HashMap，保存每一个备选电影的增强减弱因子
    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 )

//      相似度矩阵 在 离线计算中 存储时，筛选条件为 > 0.6
//      这里进一步 筛选
      if(simScore > 0.7){
        //* 打分 => 备选电影的基础推荐得分
//        ArrayBuffer +=  运算符重载 第一个()为 +=()的函数() 第二个() 为 元组
        scores += ( (candidateMovie, simScore * userRecentlyRating._2) )
//        统计 增强因子 和 减弱因子
        if (userRecentlyRating._2 > 3) {
          increMap(candidateMovie) = increMap.getOrDefault(candidateMovie, 0) + 1 //设置默认值
        } else {
          decreMap(candidateMovie) = decreMap.getOrDefault(candidateMovie, 0) + 1
        }
      }
    }
    //      除 总数
//     合并相同mid 的 groupby
    scores.groupBy(_._1).map{
//          groupBY => Map[Int, ArrayBuffer[(Int, Double)]
      case (mid, scoreList) =>
        ( mid, scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrDefault(mid, 1)) - log(decreMap.getOrDefault(mid, 1)) )
    }.toArray

  }

  /**
   * 获取两个电影之间的相似度
   * @param mid1
   * @param mid2
   * @param simMovies
   * @return
   */
  def getMoviesSimScore(mid1: Int, mid2: Int, simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Double = {
    // 用模式匹配判断是否为 空值
    simMovies.get(mid1) match {
      case Some(sims1) => sims1.get(mid2) match {
        case Some(sims2) => sims2
        case None => 0.0
      }
      case None => 0.0
    }
  }

  /**
   * 自定义 log 将 底数作为超参数
   * 超参数N 默认为 10
   */
  def log(x: Int):Double ={
    val N = 10
//    对数换底公式
    math.log(x) / math.log(N)
  }

//覆盖 更新
  def saveDataToMongoDB(uid: Int, streamRecs: Array[(Int, Double)])(implicit mongoConfig: MongoConfig): Unit ={
//    连接表
val streamRecsCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_STREAM_RECS_COLLECTION)
//    如果有 先删除
    streamRecsCollection.findAndRemove( MongoDBObject("uid" -> uid) )
//    再更新
    streamRecsCollection.insert(MongoDBObject("uid" -> uid,
      "recs" -> streamRecs.map(x => MongoDBObject("mid" -> x._1, "score" -> x._2))))
  }

}