用spark实现基于物品共现矩阵的相似度算法

发表于   |   分类于 ，   |  

本文分享一下基于共现矩阵的推荐算法实现

算法讲解

本文采用的是Jaccard相似系数：
$$w_{ij}=\frac {|N_i\cap N_j|}{|N_i|}$$
分母 $|N_i|$ 表示喜欢物品i的用户数，
分子 $|N_i\cap N_j|$ 表示同时喜欢物品i和物品j的用户数。
但是由于物品j可能是热门物品，很多人都喜欢，可能会导致最后结果接近于1，为了避免热门物品的影响，采用以下公式：
$$w_{ij}=\frac {|N_i\cap N_j|}{\sqrt{|N_i||N_j|}}$$

数据准备

本次仅需要用户偏好集
数据准备详细见[用spark实现基于物品属性相似度的推荐算法]

生成物品相似度矩阵

初始化spark运行环境

1
2
val conf = new SparkConf().setAppName("MatrixCal")
val sc = new SparkContext(conf)

不推荐在conf是设置master值，硬编码…

读取用户偏好集文件

1
val src = sc.textFile("hdfs://master:9000/user/hive/warehouse/pdata.db/user_weight")

处理物品属性文件

将物品属性集处理成（索引，（物品ID，属性数组））

1
2
3
4
5
//itemid1:userid1:rating1_1
val rdd = src.map{
  line => val lines = line.split(":")
    (lines(1),lines(0).toInt,lines(2).toDouble)
}.sortByKey().cache()

获取物品与物品的属性集

1
2
//(用户ID,物品ID)
val user_rdd2 = rdd.map(f=>(f._1,f._2))

计算物品间的共现次数

1
2
3
4
//（（物品AID，物品BID），1）
val rdd2 = user_rdd2.join(user_rdd2).map(f => (data._2,1))
//统计相同的物品对出现的次数((（物品AID，物品BID）,同时出现的总数))
val rdd3  = rdd2.reduceByKey(_+_)

计算物品间的相似度

采用Jaccard相似系数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
//对角线上((4,4),4) ((10,10),6)
val rdd4 = rdd3.filter(f => f._1._1 == f._1._2)
//非对角线上的矩阵
val rdd5 = rdd3.filter(f => f._1._1 != f._1._2)
//(4,((4,10,2),4))
val rdd6 = rdd5.map(f => (f._1._1,(f._1._1,f._1._2,f._2))).join(rdd4.map(f => (f._1._1,f._2) ))
//(10,(4,10,2,5))
val rdd7 = rdd6.map(f => (f._2._1._2,(f._2._1._1,f._2._1._2,f._2._1._3,f._2._2)))
//(10,((4,10,3,4),6))
val rdd8 = rdd7.join(rdd4.map(f => (f._1._1,f._2)))
//(4,10,3,4,6)  f._3为同时喜欢f._1,f._2的用户数，f._4,f._5为喜欢f._1,f._2的用户数
val rdd9 = rdd8.map(f => (f._2._1._1,f._2._1._2,f._2._1._3,f._2._1._4,f._2._2))
//
val rdd10 = rdd9.map(f=> (f._1, f._2, (f._3 / sqrt(f._4 * f._5)) ))
rdd10.foreach(println)

热评文章