大数据技术之电商推荐系统(8) | 实时推荐服务建设

布满荆棘的人生 2022-12-22 08:59 404阅读 0赞

### 目录 ###

*  一.实时推荐服务
 *  二.实时推荐模型和代码框架
 *   *  实时推荐模型算法设计
     *  实时推荐模块框架
 *  三.实时推荐算法的实现
 *   *  获取用户的K次最近评分
     *  获取当前商品最相似的K个商品
     *  商品推荐优先级计算
     *  将结果保存到mongoDB
 *  四.查看效果

# 一.实时推荐服务 #

实时计算与离线计算应用于推荐系统上最大的不同在于实时计算推荐结果应该反映最近一段时间用户近期的偏好，而离线计算推荐结果则是根据用户从第一次评分起的所有评分记录来计算用户总体的偏好。  
      用户对物品的偏好随着时间的推移总是会改变的。比如一个用户u 在某时刻对商品p 给予了极高的评分，那么在近期一段时候，u 极有可能很喜欢与商品p 类似的其他商品；而如果用户u 在某时刻对商品q 给予了极低的评分，那么在近期一段时候，u 极有可能不喜欢与商品q 类似的其他商品。所以对于实时推荐，当用户对一个商品进行了评价后，用户会希望推荐结果基于最近这几次评分进行一定的更新，使得推荐结果匹配用户近期的偏好，满足用户近期的口味。  
      如果实时推荐继续采用离线推荐中的ALS 算法，由于算法运行时间巨大，不具有实时得到新的推荐结果的能力；并且由于算法本身的使用的是评分表，用户本次评分后只更新了总评分表中的一项，使得算法运行后的推荐结果与用户本次评分之前的推荐结果基本没有多少差别，从而给用户一种推荐结果一直没变化的感觉，很影响用户体验。  
      另外，在实时推荐中由于时间性能上要满足实时或者准实时的要求，所以算法的计算量不能太大，避免复杂、过多的计算造成用户体验的下降。鉴于此，推荐精度往往不会很高。实时推荐系统更关心推荐结果的动态变化能力，只要更新推荐结果的理由合理即可，至于推荐的精度要求则可以适当放宽。  
      所以对于实时推荐算法，主要有两点需求：  
（1）用户本次评分后、或最近几个评分后系统可以明显的更新推荐结果；  
（2）计算量不大，满足响应时间上的实时或者准实时要求；

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

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

当用户u 对商品p 进行了评分，将触发一次对u 的推荐结果的更新。由于用户u 对商品p 评分，对于用户u 来说，他与p 最相似的商品们之间的推荐强度将发生变化，所以选取与商品p 最相似的K 个商品作为候选商品。  
每个候选商品按照“推荐优先级”这一权重作为衡量这个商品被推荐给用户u 的优先级。  
这些商品将根据用户u 最近的若干评分计算出各自对用户u 的推荐优先级，然后与上次对用户u 的实时推荐结果的进行基于推荐优先级的合并、替换得到更新后的推荐结果。  
具体来说：  
首先，获取用户u 按时间顺序最近的K 个评分，记为RK；获取商品p 的最相似的K 个商品集合，记为S；  
然后，对于每个商品qS ，计算其推荐优先级，计算公式如下：  
![在这里插入图片描述][20201118214730881.png_pic_center]  
其中：  
表示用户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）的商品个数；

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

我们在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/scala/com.donglin.streaming/OnlineRecommender.scala

# 三.实时推荐算法的实现 #

实时推荐算法的前提：  
1.在Redis集群中存储了每一个用户最近对商品的K次评分。实时算法可以快速获取。  
2.离线推荐算法已经将商品相似度矩阵提前计算到了MongoDB中。  
3.Kafka已经获取到了用户实时的评分数据。  
算法过程如下：  
实时推荐算法输入为一个评分<userId, productId, rate, timestamp>，而执行的核心内容包括：获取userId 最近K 次评分、获取productId 最相似K 个商品、计算候选商品的推荐优先级、更新对userId 的实时推荐结果。

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

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

import scala.collection.JavaConversions._
      //返回的值是一个数组Array[(productId,score)]
      def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis):Array[(Int,Double)] = { 
        //从redis中用户的评分队列里获取评分数据 list键名为uid:USERID，值格式是PRODUCTID:SCORE
        jedis.lrange( "userId:" + userId.toString,0,num)
          .map{ item =>
            val attr = item.split("\\:")
            (attr(0).trim.toInt,attr(1).trim.toDouble)
          }
          .toArray
      }

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

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

//获取当前商品的相似列表，并过滤掉用户已经评分过的，作为备选列表
      def getTopSimProducts(num: Int,
                            productId: Int,
                            userId: Int,
                            simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                           (implicit mongoConfig: MongoConfig):Array[Int] = { 
        //从广播变量相似度矩阵中拿到当前商品的相似度列表
        val allSimProducts = simProducts(productId).toArray
    
        //获取用户已经评分过的商品，过滤掉，排序输出
        val ratingCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
        val ratingExsit = ratingCollection.find(MongoDBObject("userId"->userId))
          .toArray
          .map{ item=> //只需要productId
            item.get("productId").toString.toInt
          }
        //从所有的相似商品中进行过滤
        allSimProducts.filter(x=> !ratingExsit.contains(x._1))
          .sortWith(_._2>_._2)
          .take(num)
          .map(x=>x._1)
      }

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

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

//计算每个备选商品的推荐得分
      def computeProductScore(candidateProducts: Array[Int],
                              userRecentlyRatings: Array[(Int, Double)],
                              simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
      :Array[(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 (canditateProduct <- candidateProducts;userRecentlyRating <- userRecentlyRatings){ 
          //从相似度矩阵中获取当前备选商品和当前已评分商品间的相似度
          val simScore = getProductsSimScore(canditateProduct,userRecentlyRating._1,simProducts)
          if (simScore > 0.4){ 
            //按照公式进行加权计算,得到基础评分
            scores += ((canditateProduct,simScore * userRecentlyRating._2))
            if (userRecentlyRating._2 > 3){ 
              increMap(canditateProduct) = increMap.getOrDefault(canditateProduct,0) + 1
            }else{ 
              decreMap(canditateProduct) = decreMap.getOrDefault(canditateProduct,0) + 1
            }
          }
        }
    
        //根据公式计算所有的推荐优先级，首先以productId做groupby
        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)
      }

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

def getProductsSimScore(product1: Int,
                              product2: Int,
                              simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
      :Double = { 
        simProducts.get(product1) match  { 
          case Some(sims) => sims.get(product2) match { 
            case  Some(score) => score
            case None => 0.0
          }
          case None => 0.0
        }
    
      }

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

//自定义log函数,以N为底
      def log(m:Int):Double ={ 
        val N = 10
        math.log(m)/math.log(N)
      }

## 将结果保存到mongoDB ##

// 写入mongodb
      def saveDataToMongoDB(userId: Int, streamRecs: Array[(Int, Double)])(implicit mongoConfig: MongoConfig): Unit ={ 
        val streamRecsCollection = 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)) ) )
      }

最终代码

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
    
    //定义一个助手对象，建立到redis和mongodb的连接
    object ConnHelper extends Serializable{ 
      //懒变量第一，使用的时候才初始化
      lazy val jedis = new Jedis("hadoop12")
      lazy val mongoClient = MongoClient(MongoClientURI("mongodb://hadoop12: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_PRODUCTS_NUM = 20
    
      def main(args: Array[String]): Unit = { 
        val config = Map(
          "spark.cores" -> "local[*]",
          "mongo.uri" -> "mongodb://hadoop12:27017/recommender",
          "mongo.db" -> "recommender",
          "kafka.topic" -> "recommender"
        )
        //创建spark conf
        val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("OnlineRecommender")
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
        val sc = spark.sparkContext
        val ssc = new StreamingContext(sc,Seconds(2))
    
        import spark.implicits._
        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
          //为了后续查询相似度方便，把数据转换成map形式 Map[Int, Map[Int,Double]]
          .map{ items =>
            (items.productId,items.recs.map(x=>(x.productId,x.score)).toMap)
          }
          .collectAsMap()
    
        //定义广播变量
        val simProductsMatrixBC = sc.broadcast(simProductsMatrix)
    
        //创建kafka配置参数
        val kafkaParam = Map(
          "bootstrap.servers" -> "hadoop12:9092",
          "key.deserializer" -> classOf[StringDeserializer],
          "value.deserializer" -> classOf[StringDeserializer],
          "group.id" -> "recommender",
          "auto.offset.reset" -> "latest"
        )
    
        //创建一个DStream
        val kafkaStream = KafkaUtils.createDirectStream[String,String](ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String,String](Array(config("kafka.topic")),kafkaParam)
        )
    
        //对kafkaStream进行处理，产生评分流 userId|productId|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 (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.从相似度矩阵中获取当前商品最相似的商品列表，作为备选列表,保存成一个数组Array[productId]
              val candidateProducts = getTopSimProducts(MAX_USER_RATING_NUM,productId,userId,simProductsMatrixBC.value)
    
              //3.计算每个备选商品的推荐优先级，得到当前用户的实时推荐列表,保存成Array[(productId,score)]
              val streamRecs = computeProductScore(candidateProducts,userRecentlyRatings,simProductsMatrixBC.value)
    
              //4.把推荐列表保存到mongodb
              saveDataToMongoDB(userId,streamRecs)
    
          }
        }
        //启动streaming
        ssc.start()
        print("streaming started!")
        ssc.awaitTermination()
    
      }
    
      import scala.collection.JavaConversions._
      //返回的值是一个数组Array[(productId,score)]
      def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis):Array[(Int,Double)] = { 
        //从redis中用户的评分队列里获取评分数据 list键名为uid:USERID，值格式是PRODUCTID:SCORE
        jedis.lrange( "userId:" + userId.toString,0,num)
          .map{ item =>
            val attr = item.split("\\:")
            (attr(0).trim.toInt,attr(1).trim.toDouble)
          }
          .toArray
      }
    
      //获取当前商品的相似列表，并过滤掉用户已经评分过的，作为备选列表
      def getTopSimProducts(num: Int,
                            productId: Int,
                            userId: Int,
                            simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                           (implicit mongoConfig: MongoConfig):Array[Int] = { 
        //从广播变量相似度矩阵中拿到当前商品的相似度列表
        val allSimProducts = simProducts(productId).toArray
    
        //获取用户已经评分过的商品，过滤掉，排序输出
        val ratingCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
        val ratingExsit = ratingCollection.find(MongoDBObject("userId"->userId))
          .toArray
          .map{ item=> //只需要productId
            item.get("productId").toString.toInt
          }
        //从所有的相似商品中进行过滤
        allSimProducts.filter(x=> !ratingExsit.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.immutable.Map[Int, Double]])
      :Array[(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 (canditateProduct <- candidateProducts;userRecentlyRating <- userRecentlyRatings){ 
          //从相似度矩阵中获取当前备选商品和当前已评分商品间的相似度
          val simScore = getProductsSimScore(canditateProduct,userRecentlyRating._1,simProducts)
          if (simScore > 0.4){ 
            //按照公式进行加权计算,得到基础评分
            scores += ((canditateProduct,simScore * userRecentlyRating._2))
            if (userRecentlyRating._2 > 3){ 
              increMap(canditateProduct) = increMap.getOrDefault(canditateProduct,0) + 1
            }else{ 
              decreMap(canditateProduct) = decreMap.getOrDefault(canditateProduct,0) + 1
            }
          }
        }
    
        //根据公式计算所有的推荐优先级，首先以productId做groupby
        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 getProductsSimScore(product1: Int,
                              product2: Int,
                              simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
      :Double = { 
        simProducts.get(product1) match  { 
          case Some(sims) => sims.get(product2) match { 
            case  Some(score) => score
            case None => 0.0
          }
          case None => 0.0
        }
    
      }
      //自定义log函数,以N为底
      def log(m:Int):Double ={ 
        val N = 10
        math.log(m)/math.log(N)
      }
      // 写入mongodb
      def saveDataToMongoDB(userId: Int, streamRecs: Array[(Int, Double)])(implicit mongoConfig: MongoConfig): Unit ={ 
        val streamRecsCollection = 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)) ) )
      }
    
    }

# 四.查看效果 #

使用mongodb查询  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]  
在redis添加数据  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]  
再增加一些数据

lpush userId:4867 250451:3.0 294209:1.0 457976:5.0 425715:5.0

![在这里插入图片描述][20201118204317906.png_pic_center]  
打开kafka，注意要先启动zookeeper  
启动生产者

bin/kafka-console-producer.sh --broker-list hadoop12:9092 --topic recommender

![在这里插入图片描述][20201120104822343.png_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]  
查看推荐的产品  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]

[20201118214730881.png_pic_center]: /images/20221120/e34fea6c9c844dbd856d0c4a08af2c73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/d9ee04a7ab03443db18ddafc2846359d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/e8f1640cc4224a25aaacc0766ce98ff5.png
[20201118204317906.png_pic_center]: /images/20221120/be0205e5abfa4ad19956dff9f8ffe324.png
[20201120104822343.png_pic_center]: /images/20221120/4ec99157f14741ef82da4e2bad60a8fb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/787cbad6e05243cba6891df186e9f714.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/b36e303c756d42a2824d64b45451f38d.png