协同过滤算法

利用用户行为数据进行推荐(协同过滤)

1、用户行为数据

用户行为数据在网站上最简单的存在形式就是日志，比如用户在电子商务网站中的网页浏览、购买、点击、评分和评论等活动。 用户行为在个性化推荐系统中一般分两种——显性反馈行为(explicit feedback)和隐性反馈 行为(implicit feedback)。显性反馈行为包括用户明确表示对物品喜好的行为。网站中收集显性反馈的主要方式就是评分和喜欢/不喜欢。隐性反馈行为指的是那些不能明确反应用户喜好 的行为。最具代表性的隐性反馈行为就是页面浏览行为。 按照反馈的明确性分，用户行为数据可以分为显性反馈和隐性反馈，但按照反馈的方向分， 又可以分为正反馈和负反馈。正反馈指用户的行为倾向于指用户喜欢该物品，而负反馈指用户的 行为倾向于指用户不喜欢该物品。在显性反馈中，很容易区分一个用户行为是正反馈还是负反馈， 而在隐性反馈行为中，就相对比较难以确定。

2、用户行为分析

在利用用户行为数据设计推荐算法之前，研究人员首先需要对用户行为数据进行分析，了解 数据中蕴含的一般规律，这样才能对算法的设计起到指导作用。

(1) 用户活跃度和物品流行度

(2) 用户活跃度和物品流行度的关系

一般认为，新用户倾向于浏览热门的物品，因为他 们对网站还不熟悉，只能点击首页的热门物品，而老用户会逐渐开始浏览冷门的物品。如果用横坐标表示用户活跃度，纵坐标表示具有某个活跃度的所有用户评过分的物品的平均流行度。图中曲线呈明显下 降的趋势，这表明用户越活跃，越倾向于浏览冷门的物品。

image.png

仅仅基于用户行为数据设计的推荐算法一般称为协同过滤算法。学术界对协同过滤算法进行了深入研究，提出了很多方法，比如基于邻域的方法(neighborhood-based)、隐语义模型 (latent factor model)、基于图的随机游走算法(random walk on graph)等。在这些方法中， 最著名的、在业界得到最广泛应用的算法是基于邻域的方法，而基于邻域的方法主要包含下面两种算法。

基于用户的协同过滤算法：这种算法给用户推荐和他兴趣相似的其他用户喜欢的物品

基于物品的协同过滤算法：这种算法给用户推荐和他之前喜欢的物品相似的物品

3、基于领域的算法

基于邻域的算法是推荐系统中最基本的算法，该算法不仅在学术界得到了深入研究，而且在 业界得到了广泛应用。基于邻域的算法分为两大类，一类是基于用户的协同过滤算法，另一类是 基于物品的协同过滤算法。现在我们所说的协同过滤，基本上就就是指基于用户或者是基于物品的协同过滤算法，因此，我们可以说基于邻域的算法即是我们常说的协同过滤算法

(1) 基于用户的协同过滤算法（UserCF）

基于用户的协同过滤算法的基本思想是：在一个在线个性化推荐系统中，当一个用户A需要个性化推荐 时，可以先找到和他有相似兴趣的其他用户，然后把那些用户喜欢的、而用户A没有听说过的物品推荐给A。

Ø 从上面的描述中可以看到，基于用户的协同过滤算法主要包括两个步骤。 第一步：找到和目标用户兴趣相似的用户集合。 第二步： 找到这个集合中的用户喜欢的，且目标用户没有听说过的物品推荐给目标用户。

这里，步骤1的关键是计算两个用户的兴趣相似度，协同过滤算法主要利用行为的相似度计算兴趣的相似度。给定用户u和用户v，令N(u)表示用户u曾经有过正反馈的物品集合，令N(v) 为用户v曾经有过正反馈的物品集合。那么我们可以通过以下方法计算用户的相似度：

image.png

基于余弦相似度

(2) 基于物品的协同过滤算法（itemCF）
与UserCF同理
(3) UserCF和itemCF的比较

首先我们提出一个问题，为什么新闻网站一般使用UserCF，而图书、电商网站一般使用ItemCF呢？ 首先回顾一下UserCF算法和ItemCF算法的推荐原理。UserCF给用户推荐那些和他有共同兴 趣爱好的用户喜欢的物品，而ItemCF给用户推荐那些和他之前喜欢的物品类似的物品。从这个算 法的原理可以看到，UserCF的推荐结果着重于反映和用户兴趣相似的小群体的热点，而ItemCF 的推荐结果着重于维系用户的历史兴趣。换句话说，UserCF的推荐更社会化，反映了用户所在的小型兴趣群体中物品的热门程度，而ItemCF的推荐更加个性化，反映了用户自己的兴趣传承。 在新闻网站中，用户的兴趣不是特别细化，绝大多数用户都喜欢看热门的新闻。个性化新闻推荐更加强调抓住 新闻热点，热门程度和时效性是个性化新闻推荐的重点，而个性化相对于这两点略显次要。因 此，UserCF可以给用户推荐和他有相似爱好的一群其他用户今天都在看的新闻，这样在抓住热 点和时效性的同时，保证了一定程度的个性化。同时，在新闻网站中，物品的更新速度远远快于新用户的加入速度，而且 对于新用户，完全可以给他推荐最热门的新闻，因此UserCF显然是利大于弊。

但是，在图书、电子商务和电影网站，比如亚马逊、豆瓣、Netflix中，ItemCF则能极大地发 挥优势。首先，在这些网站中，用户的兴趣是比较固定和持久的。一个技术人员可能都是在购买 技术方面的书，而且他们对书的热门程度并不是那么敏感，事实上越是资深的技术人员，他们看 的书就越可能不热门。此外，这些系统中的用户大都不太需要流行度来辅助他们判断一个物品的 好坏，而是可以通过自己熟悉领域的知识自己判断物品的质量。因此，这些网站中个性化推荐的 任务是帮助用户发现和他研究领域相关的物品。因此，ItemCF算法成为了这些网站的首选算法。 此外，这些网站的物品更新速度不会特别快，一天一次更新物品相似度矩阵对它们来说不会造成 太大的损失，是可以接受的。同时，从技术上考虑，UserCF需要维护一个用户相似度的矩阵，而ItemCF需要维护一个物品 相似度矩阵。从存储的角度说，如果用户很多，那么维护用户兴趣相似度矩阵需要很大的空间， 同理，如果物品很多，那么维护物品相似度矩阵代价较大

下表是对二者的一个全面的表较：

image.png

4、算法的实现如下：（spark）

package cf

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import breeze.math
import breeze.numerics.{pow,sqrt}
import org.apache.spark.sql.functions._

object CFUserTest {
  case class U(userid:String, itemid:String, rating:String)
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("CFUserTest")
      .master("local")
      .config("spark.sql.warehouse.dir","D:\\hadoop\\sparkTest\\data\\u.data")
      .getOrCreate()
    val data = spark.sparkContext.textFile("D:\\hadoop\\sparkTest\\data\\u.data")
    userBaseCF(data,spark)
//计算用户之间的相似度：
def userBaseCF(data:RDD[String],spark:SparkSession) : Unit ={
    import spark.implicits._
    val df_data = data.map(line => line.toString.split("\t")).map(line => U(line(0).trim,line(1).trim,line(2).trim))
      .toDF("userid","itemid","rating").select("userid","itemid","rating")

    val df_data_t = data.map(line => line.toString.split("\t")).map(line => U(line(0).trim,line(1).trim,line(2).trim))      .toDF("userid_1","itemid_1","rating_1").select("userid_1","itemid_1","rating_1")
    df_data.show(false)
    //计算每个用户用对所有item打分的平方和，再开根号
    val userScoreSum = df_data.rdd.map(x => (x(0).toString,x(2).toString)).groupByKey()
      .mapValues(x => sqrt(x.toArray.map(r => pow(r.toDouble,2)).sum))
    //将rdd userScoreSum  转换为DataFrame,并重新命名列名
    val df_user_sum =userScoreSum.toDF("user_id_sum","rating_sqrt_sum").select("user_id_sum","rating_sqrt_sum")
    df_user_sum.show()

    //对同一个item打分的用户进行join，并进行过滤，因为join是做了一个笛卡尔积，数据存在重复
    val df_decare = df_data.join(df_data_t,df_data("itemid")===df_data_t("itemid_1"))
      .filter("cast(userid as long) < cast(userid_1 as long)")
    df_decare.show()
    //定义一个udf计算两个用户对同一个item打分的乘积
    val product_udf = udf((r1:Int,r2:Int) => r1.toDouble*r2.toDouble)
    //调用udf函数，添加一列rating_product
    val df_product = df_decare.withColumn("rating_product",product_udf(col("rating"),col("rating_1")))
      .select("userid","userid_1","rating_product")
    df_product.show()
    //以"userid","userid_1"为key进行聚合，并对value进行求和
    val df_group = df_product.groupBy("userid","userid_1").agg("rating_product"->"sum")
      .withColumnRenamed("sum(rating_product)","rating_sum_pro")
    df_group.show()
    //将df_group和df_user_sum，进行内连接，目的是将rating_sqrt_sum添加到同一个DataFrame中方便计算相似度
    val df_sim_1 = df_group.join(df_user_sum,df_group("userid")===df_user_sum("cast (user_id_sum as string)"))
      .drop("user_id_sum")
    df_sim_1.show()

    //与上一步作用相同
    val df_user_sum1 = df_user_sum.withColumnRenamed("rating_sqrt_sum","rating_sqrt_sum1")
    val df_sim = df_sim_1.join(df_user_sum1,df_sim_1("userid")===df_user_sum1("userid_sum"))
      .drop("userid_sum")
    df_sim.show()
    //定义udf计算用户之间的相似度（根据cos距离计算）
    val sim_udf = udf((pro:Double,s1:Double,s2:Double) => pro/(s1.toDouble * s2.toDouble))
    val df_res = df_sim.withColumn("sim",sim_udf(col("rating_sum_pro"),col("rating_sqrt_sum"),col("rating_sqrt_sum1")))
      .select("userid","userid_1","sim")
    df_res.show()
 // 获取相似物品集合：
  def simUserItem(df_res:DataFrame,df:DataFrame,spark:SparkSession):DataFrame={
     
    import spark.implicits._
    //2.1 以user_id为key进行聚合，对value值中相似度值进行降序排列，并取前10个，然后对value进行行转列
    val df_nsim = df_res.rdd.map(x=>(x(0).toString,(x(1).toString,x(2).toString)))
      .groupByKey()
      .mapValues{x=>
        x.toArray.sortWith((x,y)=>x._2>y._2).slice(0,10)
      }.flatMapValues(x=>x).toDF("user_id","user_id1_sim")
      .selectExpr("user_id","user_id1_sim._1 as user_id1","user_id1_sim._2 as sim")
   
    val df_user_item1 = df.rdd.map(x=>(x(0).toString,x(1).toString+"_"+x(2).toString))
      .groupByKey().mapValues{x=>
      x.toArray
    }.toDF("user_id_gen_item","item_rating_array")

    val df_user_item2 = df_user_item1.withColumnRenamed("item_rating_array","item_rating_array1")

    //2.2分别为user_id和user_id1携带items进行过滤
    val df_gen_item_tmp = df_nsim.join(df_user_item1,
      df_nsim("user_id")===df_user_item1("user_id_gen_item"))
      .drop("user_id_gen_item")
    val df_gen_item = df_gen_item_tmp.join(df_user_item2,
      df_gen_item_tmp("user_id1")===df_user_item2("user_id_gen_item"))
      .drop("user_id_gen_item")

    //2.3用一个udf过滤user_id1:item2中被user_id打过分的item，其中user_id和user_id1是相似用户
    val filter_udf = udf{(items1:Seq[String],items2:Seq[String])=>
      val fMap = items1.map{x=>val l = x.split("_")
        (l(0),l(1))
      }.toMap
      items2.filter{x=>
        val l = x.split("_")
        fMap.getOrElse(l(0),"-")=="-"
      }
    }
    val df_filter_item = df_gen_item.withColumn("filtered_item",
      filter_udf(col("item_rating_array"),col("item_rating_array1")))
        .select("user_id","sim","filtered_item")
    //2.4公式计算 相似度*rating
    val sim_rating_udf = udf{(sim:Double,filter_item:Seq[String])=>
      filter_item.map{x=>val l=x.split("_")
        l(0)+"_"+l(1).toDouble*sim
      }
    }
    val itemSimRating = df_filter_item
      .withColumn("item_product",sim_rating_udf(col("sim"),col("filtered_item")))
      .select("user_id","item_product")

    val df_user_item_score = itemSimRating.select(itemSimRating("user_id"),explode(itemSimRating("item_product")))
      .toDF("user_id","item_product")
      .selectExpr("user_id","split(item_product,'_')[0] as item_id",
        "cast(split(item_product,'_')[1] as double) as score")
    df_user_item_score