在传统的推荐模型中，简单理解推荐算法召回部分的核心原理无非就是将特征以不同形式进行组织，并按照距离求解算法计算用户或物品间的相似度，那么在开始之前我们需要先了解一些常用的距离求解算法，这些算法在接下来的源码实现部分会用到

场景	算法	优势
基于客户需求的推荐	分类模型（逻辑回归、神经网络）	对当下用户行为进行意图识别
基于购物篮的推荐	关联规则	提升商品活力，挖掘用户潜在购买力
基于物品相似性的推荐	基于Item的协同过滤	分析物品潜在相似性，帮组用户快速找到想要的商品
基于用户相似性的推荐	基于User的协同过滤、KNN	形成圈子文化，发现用户潜在兴趣
基于内容相似性的推荐	TFIDF、SVD	算法可解释性强，为用户提供更多相似商品的选择
市场细分	K-means	物以类聚、人以群分

常用推荐算法有以上几种，本节我们以最经典的协调过滤进行讲解：

1、距离算法

同现矩阵

• 计算公式

• 通过案例说明
  N(i)与N(j)分别表示喜欢 i 物品的人数与喜欢 j 物品的人数，上述公式大致意思是求解喜欢物品 i 的人中又同时喜欢物品 j 的占比是多少，比值越大越能说明两个物品的关联度高，那么当其它用户去购买物品 i 时将在很大程度上喜欢物品 j ；不过需要注意的时如果物品 j 是一个热门物品，那么很多人都会喜欢物品 j，极端情况下所有喜欢物品 i 的用户都喜欢物品 j , 那么计算出的物品 i 与物品 j 就是高度相似的，为了避免热门物品的影响，在分母上对热门物品进行了惩罚，当N(j)很大时，相识度就会很低。

另外该方式还有另一个优势，那就是在计算相似度时不需要额外收集评分数据

欧几里得距离

• 计算公式

• 通过案例说明
  该方法用来计算N维空间内两点之间的距离，当N为2时欧几里得距离即平面内两点之间的距离，假设用户1与用户2分别对物品1、物品2、物品3与物品4进行的评价：

那么根据欧几里得公式将得到两个用户的相似度为：

• 通过sql实现欧氏距离

with tb1 as (
SELECT u_i as user_id, pid id, score as rating from tab   
) , 
tb2 as 
(SELECT u_i as user_id, pid id, score   as rating from tab  )

select title,recommend,sim from (
select sim_tab.*, c.title as recommend from (
sELECT item_i, item_j, 1/(1+sqrt(sum(item_ij))) sim from (
    sELECT tb1.id as item_i, tb2.id as item_j, pow(tb1.rating-tb2.rating, 2)as item_ij from tb1   LEFT join tb2   on tb1.user_id=tb2.user_id where tb1.id <> tb2.id  
) as tab  group by item_i, item_j 
) as sim_tab left join app.eqs_merchandise_model as c on sim_tab.item_j=c.id 
) as sim_tab2 left join app.eqs_merchandise_model as c on sim_tab2.item_i=c.id order by item_i,sim desc

• 通过spark实现

def euclidean2(v1: Vector, v2: Vector): Double = {
    require(v1.size == v2.size, s"SimilarityAlgorithms:Vector dimensions do not match: Dim(v1)=${v1.size} and Dim(v2)" +
      s"=${v2.size}.")

    val x = v1.toArray
    val y = v2.toArray

    euclidean(x, y)
  }

  def euclidean(x: Array[Double], y: Array[Double]): Double = {
    require(x.length == y.length, s"SimilarityAlgorithms:Array length do not match: Len(x)=${x.length} and Len(y)" +
      s"=${y.length}.")

    math.sqrt(x.zip(y).map(p => p._1 - p._2).map(d => d * d).sum)
  }
  
  def euclidean(v1: Vector, v2: Vector): Double = {
    val sqdist = Vectors.sqdist(v1, v2)
    math.sqrt(sqdist)
  }

余弦距离

• 计算公式

• 通过案例说明
  与欧几里得距离类似，也是用来求解N维空间内两个点的相似程度，不同的是余弦相似度计算的是A、B两个点与原点构成的夹角，夹角越小相似度越大；同样以上面的例子进行说明计算用户1与用户2的相似度：

• 通过sql实现余弦相似度计算

with tb1 as (
SELECT u_i as user_id, pid id, score as rating from tab   
) , 
tb2 as 
(SELECT u_i as user_id, pid id, score   as rating from tab  )

select title,recommend,sim from (
select sim_tab.*, c.title as recommend from (
sELECT item_i, item_j, sum(item_ij)/(sqrt(sum(item_i_pow)) * sqrt(sum(item_j_pow))) sim from (
    sELECT tb1.id as item_i, tb2.id as item_j, pow(tb1.rating, 2) as item_i_pow, pow(tb2.rating, 2) as item_j_pow, tb1.rating * tb2.rating as item_ij from tb1   LEFT join tb2   on tb1.user_id=tb2.user_id where tb1.id <> tb2.id  
) as tab  group by item_i, item_j 
) as sim_tab left join app.eqs_merchandise_model as c on sim_tab.item_j=c.id 
) as sim_tab2 left join app.eqs_merchandise_model as c on sim_tab2.item_i=c.id order by item_i,sim desc

皮尔逊系数

• 计算公式

• 通过案例说明
  皮尔逊相关系数是一个介于-1和1之间的数，它度量 两个一一对应数列之间的线性相关程度。也就是说，它表示两个数列中对应数字一起增大或者一起减小的可能性。它度量数字一起按比例改变的倾向性，也就是说两个数列中的数字存在一个大致的线性关系。当该倾向性强时，相关值趋于1。当相关性很弱时，相关值趋于0。在负相关的情况下一个序列的值很高而另一个序列的值低，相关性就低。假设用户3喜欢 物品1 和 物品4，评分分别为1、3，那么用户1与用户3的皮尔逊系数为：

尽量皮尔逊系数在很多推荐算法中进行了应用，但是它也有一些明显的不足之处，例如两个看过200部相同的电影的用户，即便他们给出的评分偶尔不一致，但可能要比两个看过相同两部且评分一致的用户更相似。

• 通过spark实现

def pearsonCorrelationSimilarity(arr1: Array[Double], arr2: Array[Double]): Double = {
    require(arr1.length == arr2.length, s"SimilarityAlgorithms:Array length do not match: Len(x)=${arr1.length} and Len(y)" +
      s"=${arr2.length}.")

    val sum_vec1 = arr1.sum
    val sum_vec2 = arr2.sum

    val square_sum_vec1 = arr1.map(x => x * x).sum
    val square_sum_vec2 = arr2.map(x => x * x).sum

    val zipVec = arr1.zip(arr2)

    val product = zipVec.map(x => x._1 * x._2).sum
    val numerator = product - (sum_vec1 * sum_vec2 / arr1.length)

    val dominator = math.pow((square_sum_vec1 - math.pow(sum_vec1, 2) / arr1.length) * (square_sum_vec2 - math.pow(sum_vec2, 2) / arr2.length), 0.5)
    if (dominator == 0) Double.NaN else numerator / (dominator * 1.0)
  }

2、召回算法

协同过滤

基于用户的协同过滤

• 找到和当前用户相近的一批用户
• 这批用户看过，但当前用户没有看过的商品评分乘以这个用户与当前用户的相似度分值，得到当前用户对新商品的预测分
• 将相同新商品预测分进行累加
• 新商品列表按照预测分倒序排列，取Top推荐给当前用户

通过案例进行说明：

User 1 看过 Item 1 和 Item 2，而 User 3 和 User 4 也看过 Item 1 和 Item 2，那么 User 1 和 User 3、User 4 就是相似用户。这样一来，如果 User 3 和 User 4 还分别看过 Item 3 和 Item 4，我们就可以将 Item 3 和 Item 4 都推荐给 User 1 了。但是这里有个问题，我们并没有预测User1对 Item 3 和 Item 4的偏好程度，所以也就不清楚User1更喜欢哪个。

基于上面的表格，我们可以把Item1到Item4看做是一个4维的空间向量，那么，每个用户可以认为是这4维空间中的一个向量，每一维的向量值就是当前用户对该物品的评分。n维空间求解向量相似度就可以用到我们上面说的余弦距离公式了。

分别计算与User1相似度：

预测User1对新商品Item3、Item4的偏好程度：

Item 3 的推荐打分是：1 * 0.73=0.73（User3对Item3 的喜好度 * User3 和 User1 的相似度）
Item 4 的推荐打分是：2 * 0.54 = 1.08（User4对Item4 的喜好度 * User4 和 User1 的相似度）

基于物品的协同过滤

• 基于用户物品评价矩阵，计算物品相似度矩阵
• 将物品相似度矩阵与当前用户的物品评分矩阵相乘
• 将新的矩阵按照预测评分进行倒序排列，取Top推荐给当前用户

通过案例进行说明：

为了便于大家理解，这里将上面的表格进行了行列对换，这里我们假设以用户作为维度，物品作为向量，求解两个物品在n维空间中的相似程度。
计算方式和上面计算相似用户的步骤是一致的，从上面表格我们大致可以看出Item3与Item1、Item2都相似，且User1曾经对Item1、Item2进行了评分，那么User1对Item3的偏好程度为：

Item3的推荐打分 = Item3与Item1的相似度 * User1对Item1的评分 + Item3与Item2的相似度 * User1对Item2的评分

这里我们假设已完成物品相似矩阵的计算，那么结合当前用户的评分列表就可以求出当前用户对推荐物品的偏好程度：

依次为物品相似矩阵、用户评分列表、偏好分值

协同算法思考

• 用户行为发生时间距离当前时间越近，越能反应用户的兴趣
• 相近两个行为更能反应出元素之间的相似性

基于以上假设，如何对协同过滤公式进行优化，增加如下时间衰减因子

1/(1 + α*(data[u][i][time] - data[v][i][time]))

1/(1 + α*(data[u][i][time] - data[u][j][time]))

3、 基于Spark的实现

下面通过余弦距离计算物品间相似度，关于欧式距离、同现矩阵等方式大家可以尝试修改步骤3来实现

 
  /**
   * 余弦相似度矩阵计算.
   * T(x,y) = ∑x(i)y(i) / sqrt(∑(x(i)*x(i))) * sqrt(∑(y(i)*y(i)))
   *  
   *  MovieLens 【数据地址：https://grouplens.org/datasets/movielens/】（1M、10M、20M 共三个数据集） 
   */

   // 1 数据准备
    val user_item_df = spark.read.options(Map(("delimiter",","),("header","true"))).csv("/tmp/ml-latest-small/ratings.csv")

    val user_ds1 = user_item_df.groupBy("userId").agg(collect_set(concat_ws(":","movieId","rating")).as("item_set"))

    // 2 物品:物品，上三角数据
    val user_ds2 = user_ds1.flatMap { row =>
      val itemlist = row.getAs[scala.collection.mutable.WrappedArray[String]](1).toArray.sorted
      val result = new ArrayBuffer[(String, String, Double, Double)]()
      for (i <- 0 to itemlist.length - 2) {
        for (j <- i + 1 to itemlist.length - 1) {
          result += ((itemlist(i).split(":")(0), itemlist(j).split(":")(0), itemlist(i).split(":")(1).toDouble, itemlist(j).split(":")(1).toDouble))
        }
      }
      result
    }.withColumnRenamed("_1", "itemidI").withColumnRenamed("_2", "itemidJ").withColumnRenamed("_3", "scoreI").withColumnRenamed("_4", "scoreJ")

    // 3 按照距离公式求解相似度
    // x*y = ∑x(i)y(i)
    // |x|^2 = ∑(x(i)*x(i))
    // |y|^2 = ∑(y(i)*y(i))
    // result = x*y / sqrt(|x|^2) * sqrt(|y|^2)
    val user_ds3 = user_ds2.
      withColumn("cnt", lit(1)).
      groupBy("itemidI", "itemidJ").
      agg(sum(($"scoreI" * $"scoreJ")).as("sum_xy"),
        sum(($"scoreI" * $"scoreI")).as("sum_x"),
        sum(($"scoreJ" * $"scoreJ")).as("sum_y")).
        withColumn("result", $"sum_xy" / (sqrt($"sum_x") * sqrt($"sum_y")))

    // 4 上、下三角合并
    val user_ds8 = user_ds3.select("itemidI", "itemidJ", "result").
      union(user_ds3.select($"itemidJ".as("itemidI"), $"itemidI".as("itemidJ"), $"result"))

    val user_prefer_ds2=user_ds8.join(user_item_df, $"itemidI"===$"movieId", "inner")

    //   计算召回的用户物品得分
    val user_prefer_ds3 = user_prefer_ds2.withColumn("score", col("pref") * col("similar")).select("userid", "itemidJ", "score")
    //    user_prefer_ds3.show()
    //   得分汇总
    val user_prefer_ds4 = user_prefer_ds3.groupBy("userid", "itemidJ").agg(sum("score").as("score")).withColumnRenamed("itemidJ", "itemid")
    //    user_prefer_ds4.show()
    //   用户得分排序结果，去除用户已评分物品
    val user_prefer_ds5 = user_prefer_ds4.join(user_prefer_ds1, Seq("userid", "itemid"), "left").where("pref is null")
    //    user_prefer_ds5.show()

4、 基于python的实现

import math
user_items = {}
threshold = 5
def process_data(df):
  for v in df.itertuples():
      user_items.setdefault(v[1], {})
      user_items[v[1]].setdefault(v[2], v[-1])
    
def user_similarity_cos():
    
    W = {}
    print(len(user_items.keys()))
    c = 0
    for user1 in user_items.keys():
        #W.setdefault(user1, {})
        c += 1
        if(c % 2000 == 0):
            print(c)
         
        for user2  in user_items.keys():
            if user1 == user2:
                continue
            #W[user1].setdefault(user2, 0)
            if user2 in W.keys():
                W[user1][user2] = W[user2][user1]
                continue
            
            cross_items = user_items[user1].keys() & user_items[user2].keys()
            if len(cross_items) < threshold:
                continue
            #余弦距离 
            W.setdefault(user1, {})
            #W[user1].setdefault(user2, 0)
            sum_xy = sum([user_items[user1][v] * user_items[user2][v] for v in cross_items])
            W[user1][user2] = sum_xy/(math.sqrt(sum([user_items[user1][v] * user_items[user1][v] for v in cross_items]))*math.sqrt(sum([user_items[user2][v] * user_items[user2][v] for v in cross_items])))
    
            #欧氏距离
            #dis = sum([pow(user_items[user1][v] - user_items[user2][v],2) for v in cross_items])
            #W[user1][user2] = 1/(1+math.sqrt(dis))
            
    return W
        
def recommend_cos(user, topN):
    res = {}
    items = user_items[user].keys() 
    for u, items_score in user_items.items():
        if u == user or u not in user_simi[user].keys():
            continue
        for i in items_score.keys():
            if i not in items:
                res.setdefault(i, 0)
                res[i] += user_simi[user][u] * items_score[i]
    
    sort_val = dict(sorted(res.items(), key=lambda e: -e[1]))   
    data = {}
    cn = 0
    for i, s in sort_val.items():
        cn += 1
        if cn > topN:
            break
        data[i] = s
    return data

user_simi = user_similarity_cos()
recommend('a38a23d5774a4e57bc8174928bac17d9', 5)        

基于spark sql实现

• 构建物品相似度

sql("with tb1 as (SELECT u_i as user_id, pid id, score as rating from tab   ) , tb2 as (SELECT u_i as user_id, pid id, score   as rating from tab  )sELECT item_i, item_j, sum(item_ij)/(sqrt(sum(item_i_pow)) * sqrt(sum(item_j_pow))) sim from (sELECT tb1.id as item_i, tb2.id as item_j, pow(tb1.rating, 2) as item_i_pow, pow(tb2.rating, 2) as item_j_pow, tb1.rating * tb2.rating as item_ij from tb1   LEFT join tb2   on tb1.user_id=tb2.user_id where tb1.id <> tb2.id  ) as tab  group by item_i, item_j  ").createOrReplaceTempView("sim")

• 计算推荐列表

sql(" select user, item, preference/icn preference,tab.score from(select user, item, sum(preference) preference from( select u_i as user, item_j as item, sum(sim*score) as preference from sim as a  left join tab as b on a.item_i == b.pid  group by user,item) cc left join tab dd on cc.item ==dd.pid group by user, item ) as c left join tab on tab.u_i==c.user and tab.pid == c.item ").where("score is not null").registerTempTable("rec_tab")