电商推荐系统七：基于内容的相似推荐（TF-IDF）

傷城~ 2022-10-29 01:41 218阅读 0赞

# 七、 其它形式的离线相似推荐服务 #

## 7.1 基于内容的相似推荐（TF-IDF） ##

原始数据中的tag文件，是用户给商品打上的标签，这部分内容想要直接转成评分并不容易，不过我们可以将标签内容进行提取，得到`商品的内容特征向量`，进而可以通过求取`相似度矩阵`。这部分可以与实时推荐系统直接对接，计算出与用户当前评分商品的相似商品，实现基于内容的实时推荐。为了`避免热门标签`对特征提取的影响，我们还可以通过TF-IDF算法对标签的权重进行调整，从而尽可能地接近用户偏好。

基于以上思想，加入TF-IDF算法的求取商品特征向量的核心代码如下：

然后通过商品特征向量进而求出相似度矩阵，就可以在商品详情页给出相似推荐了；通常在电商网站中，用户浏览商品或者购买完成之后，都会显示类似的推荐列表。

得到的相似度矩阵也可以为实时推荐提供基础，得到用户推荐列表。可以看出，`基于内容和基于隐语义模型，目的都是为了提取出物品的特征向量，从而可以计算出相似度矩阵`。而我们的实时推荐系统算法正是基于相似度来定义的。

package com.recom.content
    
    import org.apache.spark.SparkConf
    import org.apache.spark.ml.feature.{ HashingTF, IDF, IDFModel, Tokenizer}
    import org.apache.spark.ml.linalg.SparseVector
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{ DataFrame, SparkSession}
    import org.jblas.DoubleMatrix
    
    
    //定义样例类
    case class Product(productId:Int,name:String,imageUrl:String,categories:String,tags:String)
    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 ContentRecommender { 
      
      //定义mongodb中存储的表名
      val MONGODB_PRODUCT_COLLECTION = "Product"
      val CONTENT_PRODUCT_RECS = "ContentBasedProductRecs"
      
      def main(args: Array[String]): Unit = { 
      
        //定义基础配置的集合（可以放入配置文件，通过方法获取属性的值）
        val config = Map(
          "spark.cores"->"local[*]",
          "mongo.uri"->"mongodb://hadoop102:27017/recommender",
          "mongo.db"->"recommender"
        )
      
        //创建一个spark config
        val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("ContentRecommender")
        //创建一个spark session
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
      
        //导入隐式转换类，在DF和DS转换的过程中会使用到
        import spark.implicits._
        //通过隐式类的方法创建mongodb连接对象
        implicit val mongoConfig = MongoConfig(config("mongo.uri"),config("mongo.db"))
      
        
          //加载数据预处理，商品信息--》商品标签--》使用TF-IDF提取语义--》计算商品相似度
        val productDF = spark.read
          .option("uri",mongoConfig.uri)
          .option("collection",MONGODB_PRODUCT_COLLECTION)
          .format("com.mongodb.spark.sql")
          .load()
          .as[Product]
          .map{ 
            x=>(x.productId,x.name,x.tags.map(c=>if(c=='|') ' ' else c))
          }
          .toDF("productId","name","tags")
          .cache()
        
        
        //TODO: 使用TF-IDF提取商品特征向量
        //1.实例化一个分词器，用来做分词，默认按照空格分
        val tokenizer: Tokenizer = new Tokenizer().setInputCol("tags").setOutputCol("words")
        //用分词器做转换，得到增加一个新列words的DF
        val wordsDataDF: DataFrame = tokenizer.transform(productDF)
        //wordsDataDF.show()
        
        
        //2.定义一个HashingTF工具，计算频次
        val hashingTF: HashingTF = new HashingTF().setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(800)
        val featurizedDataDF: DataFrame = hashingTF.transform(wordsDataDF)
    // featurizedDataDF.show(truncate = false)
        
           
        //3.定义一个IDF工具，计算TF-IDF
        val idf: IDF = new IDF().setInputCol("rawFeatures").setOutputCol("features")
        //训练一个idf模型
        val idfModel: IDFModel = idf.fit(featurizedDataDF)
        //得到新增列features的DF
        val rescaledDF: DataFrame = idfModel.transform(featurizedDataDF)
    // rescaledDF.show(truncate = false)
        
        
        //对数据进行转换，得到RDD形式的features
        val productFeatures: RDD[(Int, DoubleMatrix)] = rescaledDF.map { 
          row =>
            (row.getAs[Int]("productId"),
              row.getAs[SparseVector]("features").toArray)
        }
          .rdd
          .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",CONTENT_PRODUCT_RECS)
            .mode("overwrite")
            .format("com.mongodb.spark.sql")
            .save()
        
        spark.stop()
      }
      
      //计算余弦相似度的函数
      def consinSim(product1 :DoubleMatrix, product2: DoubleMatrix):Double={ 
        
        product1.dot(product2) / (product1.norm2() * product2.norm2() )
        
      } 
    }