电商推荐系统五：实时推荐服务

迈不过友情╰ 2022-10-28 14:26 287阅读 0赞

# 五、实时推荐服务建设 #

### 目录 ###

*  五、实时推荐服务建设
 *   *  5.1 实时推荐服务
     *  5.2 实时推荐模型和代码框架
     *   *  5.2.1 实时推荐模型算法设计
         *  5.2.2 实时推荐模块框架
     *  5.3 实时推荐算法的实现
     *   *  5.3.1 获取用户的K次最近评分
         *  5.3.2 获取当前商品最相似的K个商品
         *  5.3.3 商品推荐优先级计算
         *  5.3.4 将结果保存到mongoDB
         *  5.3.5 更新实时推荐结果
     *  5.4 附件：完整代码

## 5.1 实时推荐服务 ##

实时计算与离线计算应用于推荐系统上最大的不同在于`实时计算推荐结果应该反映最近一段时间用户近期的偏好`，而`离线计算推荐结果则是根据用户从第一次评分起的所有评分记录`来计算用户总体的偏好。

`用户对物品的偏好随着时间的推移总是会改变的`。比如一个用户u 在某时刻对商品p 给予了极高的评分，那么在近期一段时候，u 极有可能很喜欢与商品p 类似的其他商品；而如果用户u 在某时刻对商品q 给予了极低的评分，那么在近期一段时候，u 极有可能不喜欢与商品q 类似的其他商品。所以对于实时推荐，当用户对一个商品进行了评价后，用户会希望推荐结果`基于最近这几次评分`进行一定的更新，使得推荐结果匹配用户`近期的偏好，满足用户近期的口味`。

如果实时推荐继续采用离线推荐中的ALS 算法，由于算法运行时间巨大，不具有实时得到新的推荐结果的能力；并且由于算法本身的使用的是评分表，用户本次评分后只更新了总评分表中的一项，使得算法运行后的推荐结果与用户本次评分之前的推荐结果基本没有多少差别，从而给用户一种推荐结果一直没变化的感觉，很影响用户体验。

另外，在`实时推荐`中由于时间性能上要`满足实时或者准实时`的要求，所以算法的`计算量不能太大`，避免复杂、过多的计算造成用户体验的下降。鉴于此，推荐精度往往不会很高。`实时推荐系统更关心推荐结果的动态变化能力`，只要更新推荐结果的理由合理即可，至于推荐的精度要求则可以适当放宽。

所以**对于实时推荐算法，主要有两点需求**：

*  （1）**用户本次评分后、或最近几个评分后系统可以明显的更新推荐结果**；
 *  （2）**计算量不大，满足响应时间上的实时或者准实时要求**；

## 5.2 实时推荐模型和代码框架 ##

### 5.2.1 实时推荐模型算法设计 ###

当用户u 对商品p 进行了评分，将触发一次对u 的推荐结果的更新。由于用户u 对商品p 评分，对于用户u 来说，他与p 最相似的商品们之间的推荐强度将发生变化，所以选取与商品p 最相似的K 个商品作为候选商品。

每个候选商品按照“`推荐优先级`”这一权重作为衡量这个商品被推荐给用户u 的优先级。

这些商品将根据用户u `最近的若干评分`计算出各自对用户u 的推荐优先级，然后与上次对用户u 的实时推荐结果的进行基于推荐优先级的合并、替换得到更新后的推荐结果。

**具体来说**：

首先，获取用户u 按时间顺序最近的K 个评分，记为RK；获取商品p 的最相似的K 个商品集合，记为S；

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

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

**公式的意义如下**：

**首先**对于每个候选商品q，从u 最近的K 个评分中，找出与q 相似度较高（>=0.6）的u 已评分商品们，对于这些商品们中的每个商品r，将r 与q 的相似度乘以用户u 对r 的评分，将这些乘积计算平均数，作为用户u 对商品q 的评分预测即  
![在这里插入图片描述][20210206205129648.png]

**然后**，将u 最近的K 个评分中与商品q 相似的、且本身评分较高（>=3）的商品个数记为 incount，计算lgmax\{incount,1\}作为商品 q 的“增强因子”，意义在于商品q 与u 的最近K 个评分中的n 个高评分(>=3)商品相似，则商品q 的优先级被增加lgmax\{incount,1\}。如果商品 q 与 u 的最近 K 个评分中相似的高评分商品越多，也就是说n 越大，则商品q 更应该被推荐，所以推荐优先级被增强的幅度较大；如果商品q 与u 的最近K 个评分中相似的高评分商品越少，也就是n 越小，则推荐优先级被增强的幅度较小；

**而后**，将u 最近的K 个评分中与商品q 相似的、且本身评分较低（< 3）的商品个数记为 recount，计算lgmax\{recount,1\}作为商品 q 的“削弱因子”，意义在于商品q 与u 的最近K 个评分中的n 个低评分(< 3)商品相似，则商品q 的优先级被削减lgmax\{incount,1\}。如果商品 q 与 u 的最近 K 个评分中相似的低评分商品越多，也就是说n 越大，则商品q 更不应该被推荐，所以推荐优先级被减弱的幅度较大；如果商品q 与u 的最近K 个评分中相似的低评分商品越少，也就是n 越小，则推荐优先级被减弱的幅度较小；

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

而在本次为用户u 实时推荐之前的上一次实时推荐结果Rec 也是一组<商品m,m 的推荐优先级>的列表，其大小也为K：  
![在这里插入图片描述][20210206205201574.png]

接下来，将updated\_S 与本次为u 实时推荐之前的上一次实时推荐结果Rec进行基于合并、替换形成新的推荐结果NewRec：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

总之，**实时推荐算法流程流程基本如下**：

1.  `用户u 对商品p 进行了评分，触发了实时推荐的一次计算`；
2.  `选出商品p 最相似的K 个商品作为集合S`；
3.  `获取用户u 最近时间内的K 条评分，包含本次评分，作为集合RK`；
4.  `计算商品的推荐优先级，产生<qID,>集合updated_S`；
5.  `将updated_S 与上次对用户u 的推荐结果Rec 利用公式(4-4)进行合并，产生新的推荐结果NewRec；作为最终输出`。

### 5.2.2 实时推荐模块框架 ###

我们在recommender下新建子项目StreamingRecommender，引入spark、scala、mongo、redis和kafka的依赖:

<dependencies>
        
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
    
        
        
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>casbah-core_2.11</artifactId>
            <version>${casbah.version}</version>
        </dependency>
        
        <dependency>
            <groupId>org.mongodb.spark</groupId>
            <artifactId>mongo-spark-connector_2.11</artifactId>
            <version>${mongodb-spark.version}</version>
        </dependency>
    
        
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>
    
        
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
    
    </dependencies>

代码中首先定义样例类和一个连接助手对象（用于建立redis和mongo连接），并在StreamingRecommender中定义一些常量：

src/main/recom/com.atguigu.streaming/StreamingRecommender.scala

// 连接助手对象
    object ConnHelper extends Serializable{ 
      lazy val jedis = new Jedis("localhost")
      lazy val mongoClient = MongoClient(MongoClientURI("mongodb://localhost:27017/recommender"))
    }
    
    case class MongConfig(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 StreamingRecommender { 
    
      val MAX_USER_RATINGS_NUM = 20
      val MAX_SIM_PRODUCTS_NUM = 20
      val MONGODB_STREAM_RECS_COLLECTION = "StreamRecs"
      val MONGODB_RATING_COLLECTION = "Rating"
      val MONGODB_PRODUCT_RECS_COLLECTION = "ProductRecs"
    //入口方法
    def main(args: Array[String]): Unit = { 
    }
    }

实时推荐主体代码如下：

def main(args: Array[String]): Unit = { 
    
      val config = Map(
        "spark.cores" -> "local[*]",
        "mongo.uri" -> "mongodb://localhost:27017/recommender",
        "mongo.db" -> "recommender",
        "kafka.topic" -> "recommender"
      )
      //创建一个SparkConf配置
      val sparkConf = new SparkConf().setAppName("StreamingRecommender").setMaster(config("spark.cores"))
      val spark = SparkSession.builder().config(sparkConf).getOrCreate()
      val sc = spark.sparkContext
      val ssc = new StreamingContext(sc,Seconds(2))
    
      implicit val mongConfig = MongConfig(config("mongo.uri"),config("mongo.db"))
      import spark.implicits._
    
      // 广播商品相似度矩阵
      //装换成为 Map[Int, Map[Int,Double]]
      val simProductsMatrix = spark
        .read
        .option("uri",config("mongo.uri"))
        .option("collection",MONGODB_PRODUCT_RECS_COLLECTION)
        .format("com.mongodb.spark.sql")
        .load()
        .as[ProductRecs]   
        .rdd
        .map{ recs =>
          (recs.productId,recs.recs.map(x=> (x.productId,x.score)).toMap)
        }.collectAsMap()  
    
      val simProductsMatrixBroadCast = sc.broadcast(simProductsMatrix)
    
      //创建到Kafka的连接
      val kafkaPara = Map(
        "bootstrap.servers" -> "localhost:9092",
        "key.deserializer" -> classOf[StringDeserializer],
        "value.deserializer" -> classOf[StringDeserializer],
        "group.id" -> "recommender",
        "auto.offset.reset" -> "latest"
      )
    
      val kafkaStream = KafkaUtils.createDirectStream[String,String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String,String](Array(config("kafka.topic")),kafkaPara))
    
      // UID|MID|SCORE|TIMESTAMP
      // 产生评分流
      val ratingStream = kafkaStream.map{ case msg=>
        var attr = msg.value().split("\\|")
        (attr(0).toInt,attr(1).toInt,attr(2).toDouble,attr(3).toInt)
      }
    
    // 核心实时推荐算法
      ratingStream.foreachRDD{ rdd =>
        rdd.map{ case (userId,productId,score,timestamp) =>
          println(">>>>>>>>>>>>>>>>")
    
          //获取当前最近的M次商品评分
          val userRecentlyRatings = getUserRecentlyRating(MAX_USER_RATINGS_NUM,userId,ConnHelper.jedis)
    
          //获取商品P最相似的K个商品
          val simProducts = getTopSimProducts(MAX_SIM_PRODUCTS_NUM,productId,userId,simProductsMatrixBroadCast.value)
    
          //计算待选商品的推荐优先级
          val streamRecs = computeProductScores(simProductsMatrixBroadCast.value,userRecentlyRatings,simProducts)
    
          //将数据保存到MongoDB
          saveRecsToMongoDB(userId,streamRecs)
    
        }.count()
      }
    
      //启动Streaming程序
      ssc.start()
      ssc.awaitTermination()
    }

## 5.3 实时推荐算法的实现 ##

实时推荐算法的前提：

1.  在Redis集群中存储了每一个用户最近对商品的K次评分。实时算法可以快速获取。
2.  离线推荐算法已经将商品相似度矩阵提前计算到了MongoDB中。
3.  Kafka已经获取到了用户实时的评分数据。

**算法过程如下**：

实时推荐算法输入为一个评分<userId, productId, rate, timestamp>，而执行的核心内容包括：获取userId 最近K 次评分、获取productId 最相似K 个商品、计算候选商品的推荐优先级、更新对userId 的实时推荐结果。

### 5.3.1 获取用户的K次最近评分 ###

业务服务器在接收用户评分的时候，默认会将该评分情况以userId, productId, rate, timestamp的格式插入到Redis中该用户对应的队列当中，在实时算法中，只需要通过Redis客户端获取相对应的队列内容即可。

import scala.collection.JavaConversions._
    /** * 获取当前最近的M次商品评分 * @param num 评分的个数 * @param userId 谁的评分 * @return */
    def getUserRecentlyRating(num:Int, userId:Int,jedis:Jedis): Array[(Int,Double)] ={ 
      //从用户的队列中取出num个评分
      jedis.lrange("userId:"+userId.toString, 0, num).map{ item =>
        val attr = item.split("\\:")
        (attr(0).trim.toInt, attr(1).trim.toDouble)
      }.toArray
    }

### 5.3.2 获取当前商品最相似的K个商品 ###

在离线算法中，已经预先将商品的相似度矩阵进行了计算，所以每个商品productId 的最相似的K 个商品很容易获取：从MongoDB中读取ProductRecs数据，从productId 在simHash 对应的子哈希表中获取相似度前K 大的那些商品。输出是数据类型为Array\[Int\]的数组，表示与productId 最相似的商品集合，并命名为candidateProducts 以作为候选商品集合。

/** * 获取当前商品K个相似的商品 * @param num 相似商品的数量 * @param productId 当前商品的ID * @param userId 当前的评分用户 * @param simProducts 商品相似度矩阵的广播变量值 * @param mongConfig MongoDB的配置 * @return */
    def getTopSimProducts(num:Int, productId:Int, userId:Int, simProducts:scala.collection.Map[Int,scala.collection.immutable.Map[Int,Double]])(implicit mongConfig: MongConfig): Array[Int] ={ 
      //从广播变量的商品相似度矩阵中获取当前商品所有的相似商品
      val allSimProducts = simProducts.get(productId).get.toArray
      //获取用户已经观看过得商品
      val ratingExist = ConnHelper.mongoClient(mongConfig.db)(MONGODB_RATING_COLLECTION).find(MongoDBObject("userId" -> userId)).toArray.map{ item =>
        item.get("productId").toString.toInt
      }
      //过滤掉已经评分过得商品，并排序输出
      allSimProducts.filter(x => !ratingExist.contains(x._1)).sortWith(_._2 > _._2).take(num).map(x => x._1)
    }

### 5.3.3 商品推荐优先级计算 ###

对于候选商品集合simiHash和userId 的最近K 个评分recentRatings，算法代码内容如下：

/** * 计算待选商品的推荐分数 * @param simProducts 商品相似度矩阵 * @param userRecentlyRatings 用户最近的k次评分 * @param topSimProducts 当前商品最相似的K个商品 * @return */
    def computeProductScores(
    	simProducts:scala.collection.Map[Int,scala.collection.immutable.Map[Int,Doub
    	le]],userRecentlyRatings:Array[(Int,Double)],topSimProducts: Array[Int]): 
    	Array[(Int,Double)] ={ 
    
      //用于保存每一个待选商品和最近评分的每一个商品的权重得分
      val score = scala.collection.mutable.ArrayBuffer[(Int,Double)]()
    
      //用于保存每一个商品的增强因子数
      val increMap = scala.collection.mutable.HashMap[Int,Int]()
    
      //用于保存每一个商品的减弱因子数
      val decreMap = scala.collection.mutable.HashMap[Int,Int]()
    
      for (topSimProduct <- topSimProducts; userRecentlyRating <- userRecentlyRatings){ 
        val simScore = getProductsSimScore(simProducts,userRecentlyRating._1,topSimProduct)
        if(simScore > 0.6){ 
          score += ((topSimProduct, simScore * userRecentlyRating._2 ))
          if(userRecentlyRating._2 > 3){ 
            increMap(topSimProduct) = increMap.getOrDefault(topSimProduct,0) + 1
          }else{ 
            decreMap(topSimProduct) = decreMap.getOrDefault(topSimProduct,0) + 1
          }
        }
      }
    
      score.groupBy(_._1).map{ case (productId,sims) =>
        (productId,sims.map(_._2).sum / sims.length + log(increMap.getOrDefault(productId, 1)) - log(decreMap.getOrDefault(productId, 1)))
      }.toArray.sortWith(_._2>_._2)
    
    }

其中，getProductSimScore是取候选商品和已评分商品的相似度，代码如下：

/** * 获取当个商品之间的相似度 * @param simProducts 商品相似度矩阵 * @param userRatingProduct 用户已经评分的商品 * @param topSimProduct 候选商品 * @return */
    def getProductsSimScore(
    simProducts:scala.collection.Map[Int,scala.collection.immutable.Map[Int,Double]], userRatingProduct:Int, topSimProduct:Int): Double ={ 
      simProducts.get(topSimProduct) match { 
        case Some(sim) => sim.get(userRatingProduct) match { 
          case Some(score) => score
          case None => 0.0
        }
        case None => 0.0
      }
    }

而log是对数运算，这里实现为取10的对数（常用对数）：

//取10的对数
    def log(m:Int):Double ={ 
      math.log(m) / math.log(10)
    }

### 5.3.4 将结果保存到mongoDB ###

saveRecsToMongoDB函数实现了结果的保存：

/** * 将数据保存到MongoDB userId -> 1, recs -> 22:4.5|45:3.8 * @param streamRecs 流式的推荐结果 * @param mongConfig MongoDB的配置 */
    def saveRecsToMongoDB(userId:Int,streamRecs:Array[(Int,Double)])(implicit mongConfig: MongConfig): Unit ={ 
      //到StreamRecs的连接
      val streaRecsCollection = ConnHelper.mongoClient(mongConfig.db)(MONGODB_STREAM_RECS_COLLECTION)
    
      streaRecsCollection.findAndRemove(MongoDBObject("userId" -> userId))
      streaRecsCollection.insert(MongoDBObject("userId" -> userId, "recs" ->
    	streamRecs.map( x => MongoDBObject("productId"->x._1,"score"->x._2)) ))
    }

### 5.3.5 更新实时推荐结果 ###

当计算出候选商品的推荐优先级的数组updatedRecommends<productId, E>后，这个数组将被发送到Web 后台服务器，与后台服务器上userId 的上次实时推荐结果recentRecommends<productId, E>进行合并、替换并选出优先级E 前K大的商品作为本次新的实时推荐。具体而言：

*  a.合并：将updatedRecommends 与recentRecommends 并集合成为一个新的<productId, E>数组；
 *  b.替换（去重）：当updatedRecommends 与recentRecommends 有重复的商品productId 时，recentRecommends 中productId 的推荐优先级由于是上次实时推荐的结果，于是将作废，被替换成代表了更新后的updatedRecommends的productId 的推荐优先级；
 *  c.选取TopK：在合并、替换后的<productId, E>数组上，根据每个product 的推荐优先级，选择出前K 大的商品，作为本次实时推荐的最终结果。

## 5.4 附件：完整代码 ##

package com.recom.online
    
    import com.mongodb.casbah.commons.MongoDBObject
    import com.mongodb.casbah.{ MongoClient, MongoClientURI, MongoCollection}
    import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.kafka.common.serialization.StringDeserializer
    import org.apache.spark.{ SparkConf, SparkContext}
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.streaming.dstream.{ DStream, InputDStream}
    import org.apache.spark.streaming.kafka010.{ ConsumerStrategies, KafkaUtils, LocationStrategies}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import redis.clients.jedis.Jedis
    
    //定义一个连接助手对象，建立到redis和mongodb的连接
    object ConnHelper extends Serializable{ 
      //懒变量定义，使用的时候再进行初始化
      lazy val jedis = new Jedis("hadoop102")
      lazy val mongoClient = MongoClient(MongoClientURI("mongodb://hadoop102:27017/recommender"))
    }
    
    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 OnlineRecommender { 
      
      //定义常量和表名
      val MONGODB_RATING_COLLECTION = "Rating"
      val STREAM_RECS="StreamRecs"
      val PRODUCT_RECS="ProductRecs"
      val MAX_USER_RATING_NUM=20
      val MAX_SIM_PRODUCTS_NUM=20
      
      def main(args: Array[String]): Unit = { 
      
        //定义基础配置的集合（可以放入配置文件，通过方法获取属性的值）
        val config = Map(
          "spark.cores"->"local[*]",
          "mongo.uri"->"mongodb://hadoop102:27017/recommender",
          "mongo.db"->"recommender",
          "kafka.topic"->"recommender"
        )
      
        //创建一个spark config
        val sparkConf: SparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("OnlinRecommender")
        //创建一个spark session
        val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
        val sc: SparkContext = spark.sparkContext
        val ssc: StreamingContext = new StreamingContext(sc,Seconds(2))
        
        //导入隐式转换类，在DF和DS转换的过程中会使用到
        import spark.implicits._
        //通过隐式类的方法创建mongodb连接对象
        implicit val mongoConfig = MongoConfig(config("mongo.uri"),config("mongo.db"))
      
        //加载数据, 相似度矩阵，并且广播出去
        val simProductsMatrix = spark.read
          .option("uri", mongoConfig.uri)
          .option("collection", PRODUCT_RECS)
          .format("com.mongodb.spark.sql")
          .load()
            .as[ProductRecs]
            .rdd    //为了调用rdd中的collectAsMap算子
            //为了后续查询相似度方便，把数据格式转换成map形式
            .map{ 
              item=>(item.productId,item.recs.map(x=>(x.productId,x.score)).toMap)
            }
            .collectAsMap()
        
        //定义广播变量
        val simProductMatrixBC = sc.broadcast(simProductsMatrix)
        
        //创建kafka配置参数
        val kafkaParam = Map(
          "bootstrap.servers"->"hadoop102:9092",
          "key.deserializer"->classOf[StringDeserializer],
          "value.deserializer"->classOf[StringDeserializer],
          "group.id"->"recommender",
          "auto.offset.reset"->"latest"
        )
        
        //创建一个kafka DStream
        val kafkaStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, String](Array(config("kafka.topic")), kafkaParam)
        )
        //对kafkaStrream进行处理，产生评分流，userId|productId|score|timestamp
        val ratingStream: DStream[(Int, Int, Double, Int)] = kafkaStream.map {  msg =>
          val arrs: Array[String] = msg.value().split("\\|")
          (arrs(0).toInt, arrs(1).toInt, arrs(2).toDouble, arrs(3).toInt)
        }
           
        
        //核心算法部分，定义评分流的处理流程
        ratingStream.foreachRDD{ 
          rdds=>rdds.foreach{ 
            case (userId,productId,score,timestamp)=>
              println("rating data coming!>>>>>>>>>>>>>>>>>>>>>>>>>>>")
              
              //TODO:核心算法逻辑
              //1.从redis中取出当前用户的最近评分，保存成一个数组Array[(productId,score)]
              val userRecentlyRatings= getUserRecentlyRatings(MAX_USER_RATING_NUM,userId,ConnHelper.jedis)
            
              //2.从相似度矩阵中获取当前商品最相似的商品列表，作为备选列表（需要取出用户已经评论过的商品），保存成一个数组Arry[productId]
              val candidateProducts = getTopSimProducts(MAX_SIM_PRODUCTS_NUM,productId,userId,simProductMatrixBC.value)
            
              //3.计算每个备选商品的推荐优先级别，得到当前用户的实时推荐列表，保存成Arry[(productId,score)]
              val streamRecs = computeProductScore(candidateProducts,userRecentlyRatings,simProductMatrixBC.value)
            
              //4.把推荐列表保存到mongodb
              saveDataToMongoDB(userId,streamRecs)
          }
        }
      
        //启动straing
        ssc.start()
        println("streaming start collecting data!!!")
        ssc.awaitTermination()
        
      }
      
      
      /* 从redis中获取最近num次评分 */
      import scala.collection.JavaConversions._
      def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis)={ 
        //从redis中用户的评分队列中获取评分数据，list键名为uid:USERID，值格式为PRODUCTID:SCORE
        val result = jedis.lrange("userId:"+userId.toString,0,num)
          .map{ item=>{ 
             val arrs: Array[String] = item.split("\\:")
            (arrs(0).trim.toInt,arrs(1).trim.toDouble)
          }}
          .toArray
        
        result
      }
      
      
      /* 获取当前商品的相似列表，并过滤掉用户已经评过分的，作为备选列表 */
      def getTopSimProducts(num: Int, productId: Int, userId: Int,
                            simProducts: scala.collection.Map[Int, scala.collection.Map[Int, Double]])
                           (implicit mongoConfig: MongoConfig)={ 
        
        //从广播变量相似度矩阵中获取当前商品的相似度列表
        val allSimProducts = simProducts(productId).toArray
        
        //获取用户已经评过分的商品，过滤掉，排序输出
        val ratingCollection: MongoCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
        val ratingExist = ratingCollection
          .find(MongoDBObject("userId"->userId))
          .toArray
          .map{  //只需要productId
            item=>item.get("productId").toString.toInt
          }
        
        //从所有的相似商品中进行过滤
        allSimProducts.filter(x=> !ratingCollection.contains(x._1))
          .sortWith(_._2>_._2)
          .take(num)
          .map(x=>x._1)
      }
      
      
      /* 计算每个备选商品的推荐得分 */
      def computeProductScore(candidateProducts: Array[Int],
                              userRecentlyRatings: Array[(Int, Double)],
                              simProducts: scala.collection.Map[Int, scala.collection.Map[Int, Double]])={ 
        
        //定义一个长度可变数组ArrayBuffer，用于保存每一个备选商品的基础得分，（productId，score）
        val scores = scala.collection.mutable.ArrayBuffer[(Int,Double)]()
        
        //定义两个Map用于保存每个商品的高分和低分的计数器，productId->count
        val increMap = scala.collection.mutable.HashMap[Int,Int]()
        val decreMap = scala.collection.mutable.HashMap[Int,Int]()
        
        //遍历每个备选商品，计算和已评分商品的相似度
        for (candidateProduct<-candidateProducts;userRecentlyRating<-userRecentlyRatings){ 
          
          //从相似度矩阵中获取当前备选商品和当前评分商品之间的相似度
          val simScore = getProductSimScore(candidateProduct,userRecentlyRating._1,simProducts)
          if (simScore>0.4){ 
            //按照公式进行加权计算，得到基础得分
            scores+=((candidateProduct, simScore*userRecentlyRating._2))
            
            if (userRecentlyRating._2 > 3){ 
              increMap(candidateProduct)=increMap.getOrDefault(candidateProduct,0)+1
            }else{ 
              decreMap(candidateProduct)=increMap.getOrDefault(candidateProduct,0)+1
            }
          }
        }
        
        //根据公式计算所有的推荐优先级，首先以productId做group
        scores.groupBy(_._1).map{ 
          case (productId,scoreList)=>(productId,
              scoreList.map(_._2).sum/scoreList.length + log(increMap.getOrDefault(productId,1))
                              -log(decreMap.getOrDefault(productId,1)))
        }
        //返回推荐列表，按照得分排序
          .toArray
          .sortWith(_._2>_._2)
      }
      
      //获取两个商品之间的相似度
      def getProductSimScore(product1: Int, product2: Int,
                             simProducts: scala.collection.Map[Int, scala.collection.Map[Int, Double]])={ 
      
        simProducts.get(product1) match { 
          case None =>0.0
          case Some(x) => x.get(product2) match { 
            case None =>0.0
            case Some(x) =>x
          }
        }
      }
      
      //自定义log函数，以N为底
      def log(m: Int)={ 
        val N = 10
        math.log(m) / math.log(N)
      }
      
      //写入mongodb
      def saveDataToMongoDB(userId: Int, streamRecs: Array[(Int, Double)])
                           (implicit mongoConfig: MongoConfig)={ 
        
        val streamRecsCollection: MongoCollection = ConnHelper.mongoClient(mongoConfig.db)(STREAM_RECS)
        //按照userId查询并更新
        streamRecsCollection.findAndRemove(MongoDBObject("userID"->userId))
        streamRecsCollection.insert(MongoDBObject("userId"->userId,     "recs"-> streamRecs.map(x=>MongoDBObject("productId"->x._1, "score"->x._2))) )
        
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221024/e78deb6b93794ab3aaebb00bc572f37b.png
[20210206205129648.png]: /images/20221024/77f43bf045e649348420056afb53424f.png
[20210206205154358.png]: /images/20221024/67154644b3ef4fc7b354f23b088f8fc3.png
[20210206205201574.png]: /images/20221024/86590ed435344d4bb80cb053fab796f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221024/be78680bc3314647a3a0172bcbba4a09.png