一、商品相似度定义

      基于物品的协同过滤算法是业界应用最多的算法，它的思想是给用户推荐那些和他们喜欢的物品相似的物品，主要分为两个步骤：一，计算物品之间的相似度；二，根据物品相似度和用户的历史行为给用户生成推荐列表。

物品i和物品j的相似度可定义为：

其中，分母是喜欢物品i的用户数，分子是同时喜欢物品i和物品j的用户数。

   在电商网站中，用户前后的消费行为有很强的关联性，如上图所示，两个用户userA,userB在易购上面的消费记录（https://www.jianshu.com/p/00a28141521f），如果灰色方框内用户购买的商品互换，可能性是很小的。用户在较短时间内点击的商品相似度也更高，在物品与物品的相似度中加入时间衰减因子： 

二、spark计算商品相似度

1、用来测试的数据如下，格式：用户cookie，商品pid，日期dt

from pyspark.sql import Row
clickData = sc.parallelize([\
      Row(cookie=1,pid=1,dt='2018-06-11'),\
      Row(cookie=1,pid=2,dt='2018-06-11'),\
      Row(cookie=1,pid=3,dt='2018-07-02'),\
      Row(cookie=2,pid=1,dt='2018-06-15'),\
      Row(cookie=2,pid=3,dt='2018-06-11'),\
      Row(cookie=3,pid=2,dt='2018-06-20'),\
      Row(cookie=3,pid=3,dt='2018-07-02')]).toDF()

2、(cookie,pid,dt)进行自连接，获得用户同时购买的商品

pidClickDataPart = clickData.repartition('cookie').persist()
pidCartData = pidClickDataPart.join(pidClickDataPart,'cookie').toDF('cookie','dt1','pid1','dt2','pid2').filter('pid1 != pid2')

数据分区：连接操作将两个数据集中的所有建的哈希值都求出来，将哈希值相同的记录通过网络传到同一台机器上，然后在那台机器上对所有键相同的记录进行链接操作。不同节点间的网络通信如下所示

clickData通过repartition建立分区（需要进行持久化，因为spark每次用到这个dataframe时会重新进行计算，持久化可防止重复进行分区操作），spark就知道该dataframe是根据键的哈希值来分区的，在调用join是，spark会利用这一点，减少不同节点间的网络通信。

3、计算时间衰减因子

pidDiffTimeData = pidCartData.withColumn('timeDiff',fn.abs(fn.datediff(pidCartData.dt1,pidCartData.dt2)))
pidTimeWeight = pidDiffTimeData.withColumn('timeWeight', fn.exp(-pidDiffTimeData.timeDiff/fn.lit(30)))

4、计算商品之间的相似度

pidSimSum = pidTimeWeight.groupBy('pid1','pid2').sum('timeWeight').toDF('pid1','pid2','simSum')
pidCount = clickData.groupBy('pid').count()
pidCountSimSum = pidSimSum.join(pidCount, pidSimSum.pid1==pidCount.pid)
pidSimScore = pidCountSimSum.withColumn('simscore', pidCountSimSum['simSum']/pidCountSimSum['count'])\
        .select('pid1','pid2','simscore').toDF('pid1','pid2','similarity')\
        .withColumn('rank', fn.row_number().over(Window.partitionBy('pid1').orderBy(fn.desc('similarity'))))