推荐算法--基于物品的协同过滤及Spark实现

问题导读
1. 基于物品的协同过滤是什么？其核心思想是什么？
2. 使用Python如何实现基于物品的协同过滤？
3. 在大数据集群上，用Spark如何实现基于物品的协同过滤？




关注最新经典文章，欢迎关注公众号


综述
本文基于项亮大神的《推荐系统实践》介绍基于物品的协同过滤，并用python和spark分别进行了实现。
基于物品的协同过滤（item-based collaborative filtering）算法是目前业界应用最多的算法。无论是亚马逊网，还是Netflix、 Hulu、 YouTube，其推荐算法的基础都是该算法。

基于物品的协同过滤算法（简称ItemCF）给用户推荐那些和他们之前喜欢的物品相似的物品。比如，该算法会因为你购买过《数据挖掘导论》而给你推荐《机器学习》。不过， ItemCF算法并不利用物品的内容属性计算物品之间的相似度，它主要通过分析用户的行为记录计算物品之间的相似度。该算法认为，物品A和物品C具有很大的相似度是因为喜欢物品A的用户大都也喜欢物品C。
如图所示：


算法基本步骤
基于物品的协同过滤算法主要分为两步：
一、计算物品之间的相似度；
二、根据物品的相似度和用户的历史行为给用户生成推荐列表；

计算相似度
主要使用如下公式进行计算：


其中，|N(i)|是喜欢物品i的用户数，|N(j)|是喜欢物品j的用户数，|N(i)&N(j)|是同时喜欢物品i和物品j的用户数。

从上面的定义看出，在协同过滤中两个物品产生相似度是因为它们共同被很多用户喜欢，两个物品相似度越高，说明这两个物品共同被很多人喜欢。

这里面蕴含着一个假设：就是假设每个用户的兴趣都局限在某几个方面，因此如果两个物品属于一个用户的兴趣列表，那么这两个物品可能就属于有限的几个领域，而如果两个物品属于很多用户的兴趣列表，那么它们就可能属于同一个领域，因而有很大的相似度。
下是一个根据上面的程序计算物品相似度的简单例子。图中最左边是输入的用户行为记录，每一行代表一个用户感兴趣的物品集合。然后，对于每个物品集合，我们将里面的物品两两加一，得到一个矩阵。最终将这些矩阵相加得到上面的C矩阵。其中C[j]记录了同时喜欢物品i和物品j的用户数。最后，将C矩阵归一化可以得到物品之间的余弦相似度矩阵W。


生成推荐列表：
ItemCF通过如下公式计算用户u对一个物品j的兴趣：

这里N(u)是用户喜欢的物品的集合， S(j,K)是和物品j最相似的K个物品的集合， wji是物品j和i的相似度， rui是用户u对物品i的兴趣。（对于隐反馈数据集，如果用户u对物品i有过行为，即可令
rui=1。）该公式的含义是，和用户历史上感兴趣的物品越相似的物品，越有可能在用户的推荐列表中获得比较高的排名。

代码实现
根据《推荐系统实践》 给出了如下python 版本的相似度计算
[mw_shl_code=python,true]    def calc_movie_sim(self):
        for user, movies in self.trainset.items():
            for movie in movies:
                if movie not in self.movie_popular:
                    self.movie_popular[movie] = 0
                self.movie_popular[movie] += 1
        print('count movies number and pipularity succ', file=sys.stderr)

        self.movie_count = len(self.movie_popular)
        print('total movie number = %d' % self.movie_count, file=sys.stderr)

        itemsim_mat = self.movie_sim_mat
        print('building co-rated users matrix', file=sys.stderr)
        for user, movies in self.trainset.items():
            for m1 in movies:
                for m2 in movies:
                    if m1 == m2:
                        continue
                    itemsim_mat.setdefault(m1, {})
                    itemsim_mat[m1].setdefault(m2, 0)
                    itemsim_mat[m1][m2] += 1

        print('build co-rated users matrix succ', file=sys.stderr)
        print('calculating movie similarity matrix', file=sys.stderr)

        simfactor_count = 0
        PRINT_STEP = 2000000

        for m1, related_movies in itemsim_mat.items():
            for m2, count in related_movies.items():
                itemsim_mat[m1][m2] = count / math.sqrt(self.movie_popular[m1] * self.movie_popular[m2])
                simfactor_count += 1
                if simfactor_count % PRINT_STEP == 0:
                    print('calcu movie similarity factor(%d)' % simfactor_count, file=sys.stderr)
        print('calcu similiarity succ', file=sys.stderr)[/mw_shl_code]

整体效率还是比较低的。因为我们实际的业务，使用的spark进行ELT。为了存储方便，将数据存储为稀疏矩阵的形式，如下：
User movie1:ratting1 movie2:ratting2 .....

使用load_svmlight_file 导入，然后直接调用jaccard 距离计算。
[mw_shl_code=python,true]def get_data(filename):
    data = load_svmlight_file(filename)
    return data[0], data[1]


# 计算jaccard 相似度
def get_jaccard_similarity(X):
    n = X.T.shape[1]
    similarity = np.zeros([n, n])
    for i in range(n):
        v1 = X.T.toarray()
        for j in range(i + 1, n):
            v2 = X.T[j].toarray()
            sim = jaccard_similarity_score(v1, v2)
            similarity[j] = sim
            similarity[j] = sim
    return similarity[/mw_shl_code]

例子解释
用户喜欢《C++ Primer中文版》和《编程之美》两本书。然后ItemCF会为这两本书分别找出和它们最相似的3本书，然后根据公式的定义计算用户对每本书的感兴趣程度。比如， ItemCF给用户推荐《算法导论》，是因为这本书和《C++Primer中文版》相似，相似度为0.4，而且这本书也和《编程之美》相似，相似度是0.5。考虑到用户对《C++ Primer中文版》的兴趣度是1.3，对《编程之美》的兴趣度是0.9，那么用户对《算法导论》的兴趣度就是1.3 × 0.4 + 0.9×0.5 = 0.97。
过程如图：




Spark实现
因为数据都在CDH集群上，且量比较大，download到本地资源浪费比较明显，基于BitSet写了一版Spark版本计算Jaccard距离的。
使用BitSet 比较节约资源。

代码如下：
[mw_shl_code=scala,true]  /**
    * 使用BitSet 计算jaccard 距离
    */
  def computeItemJaccardSim(sc: SparkContext,
                        featurePath: String,
                        simOutPath: String): Unit ={
    val rdd = sc.textFile(featurePath)
      .map(_.split(" ", 2)(1))
      .zipWithIndex()
      .map(x => (x._2.toInt,x._1.split(" ", -1)))
      .map(x=>{
        for (i <- x._2) yield {
          (i.split("\\:")(0), x._1)
        }
      }).flatMap(x=>x)
      .map(x=>(x._1,BitSet(x._2.toString.toInt)))
      .reduceByKey(_.union(_))

    val re = rdd.cartesian(rdd).map {
      case((key0,set0),(key1,set1))=>{
        val key=key0+","+key1
        val j = (set0 &(set1)).size
        val q = set0.union(set1).size
        val re = j.toDouble/q
        key+","+re
      }
    }
    re.saveAsTextFile(simOutPath)
    println("compute jaccard sim success!")
  }[/mw_shl_code]


总结
介绍了基于物品协同过滤的基本概念，并给出了相关实现代码。代码、数据集整理在如下github 欢迎star

基于物品的协同过滤及Spark实现， 很好

学习学习