大数据技术之电商推荐系统(6) | 基于LFM的离线推荐模块

ゝ一世哀愁。 2022-12-21 06:26 344阅读 0赞

### 目录 ###

*  用户商品推荐列表
 *  商品相似度矩阵
 *  扩展写到Mysql里

# 用户商品推荐列表 #

通过ALS训练出来的Model来计算所有当前用户商品的推荐列表，主要思路如下：  
1.userId和productId做笛卡尔积，产生（userId，productId）的元组  
2.通过模型预测（userId，productId）对应的评分。  
3.将预测结果通过预测分值进行排序。  
4.返回分值最大的K个商品，作为当前用户的推荐列表。  
最后生成的数据结构如下：将数据保存到MongoDB的UserRecs表中  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]  
新建recommender的子项目OfflineRecommender，引入spark、scala、mongo和jblas的依赖：

<dependencies>
    
        <dependency>
            <groupId>org.scalanlp</groupId>
            <artifactId>jblas</artifactId>
            <version>${ jblas.version}</version>
        </dependency>
    
        
        <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-mllib_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>
    </dependencies>

同样经过前期的构建样例类、声明配置、创建SparkSession等步骤，可以加载数据开始计算模型了。  
核心代码如下：  
src/main/scala/com.donglin.offline/OfflineRecommender.scala

# 商品相似度矩阵 #

通过ALS计算商品相似度矩阵，该矩阵用于查询当前商品的相似商品并为实时推荐系统服务。  
离线计算的ALS 算法，算法最终会为用户、商品分别生成最终的特征矩阵，分别是表示用户特征矩阵的U(m x k)矩阵，每个用户由 k个特征描述；表示物品特征矩阵的V(n x k)矩阵，每个物品也由 k 个特征描述。  
V(n x k)表示物品特征矩阵，每一行是一个 k 维向量，虽然我们并不知道每一个维度的特征意义是什么，但是k 个维度的数学向量表示了该行对应商品的特征。  
所以，每个商品用V(n x k)每一行的向量表示其特征，于是任意两个商品 p：特征向量为，商品q：特征向量为之间的相似度sim(p,q)可以使用和的余弦值来表示：  
![在这里插入图片描述][20201115171635749.png_pic_center]  
数据集中任意两个商品间相似度都可以由公式计算得到，商品与商品之间的相似度在一段时间内基本是固定值。最后生成的数据保存到MongoDB的ProductRecs表中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]

val productFeatures = model.productFeatures.map{ 
          case (productId,features) => (productId,new DoubleMatrix(features))
        }
        //两两配对商品，计算余弦相似度
        val productRecs = productFeatures.cartesian(productFeatures)
            .filter{ 
              case (a,b) => a._1 != b._1
            }
        //计算余弦相似度
            .map{ 
              case(a,b) =>
                val simScore = consinSim(a._2,b._2)
                (a._1,(b._1,simScore))
            }
            .filter(_._2._2 > 0.4)
            .groupByKey()
            .map{ 
              case (productId,recs) =>
                ProductRecs(productId,recs.toList.sortWith(_._2>_._2).map(x=>Recommendation(x._1,x._2)))
            }
            .toDF()
    
        productRecs.write
          .option("uri",mongoConfig.uri)
          .option("collection",PRODUCT_RECS)
          .mode("overwrite")
          .format("com.mongodb.spark.sql")
          .save()

其中，consinSim是求两个向量余弦相似度的函数，代码实现如下：

//计算两个商品之间的余弦相似度
    def consinSim(product1: DoubleMatrix, product2:DoubleMatrix) : Double ={ 
      product1.dot(product2) / ( product1.norm2()  * product2.norm2() )
    }

最终代码实现

import org.apache.spark.SparkConf
    import org.apache.spark.mllib.recommendation.{ ALS, Rating}
    import org.apache.spark.sql.SparkSession
    import org.jblas.DoubleMatrix
    case class ProductRating(userId:Int,productId:Int,score:Double,timestamp:Int)
    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 { 
      //定义mongodb中存储表名
      val MONGODB_RATING_COLLECTION = "Rating"
    
      val USER_RECS = "UserRecs"
      val PRODUCT_RECS = "ProductRecs"
      val USER_MAX_RECOMMENDATION = 3
    
      def main(args: Array[String]): Unit = { 
        val config = Map(
          "spark.cores" -> "local[*]",
          "mongo.uri" -> "mongodb://hadoop12:27017/recommender",
          "mongo.db" -> "recommender"
        )
    
        //创建一个spark config
        val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("OfflineRecommender")
        //创建spark session
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    
        import spark.implicits._
        implicit val mongoConfig = MongoConfig(config("mongo.uri"),config("mongo.db"))
    
        //加载数据
        val ratingRDD = 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()
    
        //提取出所有用户和商品的数据集
        val userRDD = ratingRDD.map(_._1).distinct()
        val productRDD = ratingRDD.map(_._2).distinct()
    
        //TODO ：核心计算过程
        //1.训练隐语义模型
        val trainData = ratingRDD.map(x => Rating(x._1, x._2, x._3))
        //定义茉寻训练的参数 rank隐特征个数，iterations迭代次数，lambda正则化系数
        val (rank, iterations, lambda) = (10,10,0.01)
        val model = ALS.train(trainData, rank, iterations, lambda)
    
        //2.获得预测评分矩阵，得到用户的推荐列表
        // 用userRDD 和 productRDD做一个笛卡尔积，得到空的userProductsRDD
        val userProducts = userRDD.cartesian(productRDD)
        //preRating输出是这样的 Rating(3455,286997,4.401263661039027)
        val preRating = model.predict(userProducts)
    
        //从预测评分矩阵中提取得到用户列表
        val userRecs = 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()
    
        userRecs.write
            .option("uri",mongoConfig.uri)
            .option("collection",USER_RECS)
            .mode("overwrite")
            .format("com.mongodb.spark.sql")
            .save()
    
        //3.利用商品的特征向量，计算商品相似度列表
        val productFeatures = model.productFeatures.map { 
          case (productId, feature) => (productId, new DoubleMatrix(feature))
        }
        /*productFeatures输出(470839,[-0.405416876077652; 0.007109188940376043; -0.13823118805885315; 0.6249048113822937; -0.35293328762054443; -1.0173797607421875; -0.28891870379447937; 0.01371031068265438; 0.7766180038452148; 0.3023674488067627]) */
        //两两配对商品，计算余弦相似度
        val productRecs = productFeatures.cartesian(productFeatures)
          .filter { 
            case (a, b) => a._1 != b._1
          }
          //计算余弦相似度
          .map { 
            case (a, b) =>
              val simScore = consinSim(a._2, b._2)
              (a._1, (b._1, simScore))
          }
          .filter(_._2._2 > 0.4)
          .groupByKey()
          .map { 
            case (productId, recs) =>
              ProductRecs(productId, recs.toList.sortWith(_._2 > _._2).take(USER_MAX_RECOMMENDATION).map(x => Recommendation
              (x._1, x._2)))
          }
          .toDF()
    
        productRecs.write
            .option("uri",mongoConfig.uri)
            .option("collection",PRODUCT_RECS)
            .mode("overwrite")
            .format("com.mongodb.spark.sql")
            .save()
    
    
        spark.close()
      }
    
      def consinSim(product1: DoubleMatrix, product2: DoubleMatrix):Double={ 
        //模长是norm2
        product1.dot(product2)/(product1.norm2() * product2.norm2())
      }
    
    }

查看结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]

# 扩展写到Mysql里 #

遇到了报错

Exception in thread "main" java.lang.IllegalArgumentException: Can't get JDBC type for array<struct<productId:int,score:double>>

原因是mysql里面不能存储array，尽量存在redis mongodb里

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/80b097348106409593c6b44f145efdb6.png
[20201115171635749.png_pic_center]: /images/20221120/fee379c740ee4e8784f7404a4143afb2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/1fd03000978b4adaa6b6f4ea501959ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/c4181cc20f094578905b9d465e921bf3.png