Spark ML協同過濾推薦算法

一.簡介

　　協同過濾算法【Collaborative Filtering Recommendation】算法是最經典、最常用的推薦算法。該算法通過分析用戶興趣，在用戶群中找到指定用戶的相似用戶，綜合這些相似用戶對某一信息的評價，形成系統關於該指定用戶對此信息的喜好程度預測。

二.步驟

　　1.收集用戶偏好。

　　2.找到相似的用戶或物品。

　　3.計算推薦。

三.用戶評分

　　從用戶的行為和偏好中發現規律，並基於此進行推薦，所以收集用戶的偏好信息成為系統推薦效果最基礎的決定因素。

　　數據預處理：

　　　　1.減噪

　　　　　　因為用戶數據在使用過程中可能存在大量噪聲和誤操作，所以需要過濾掉這些噪聲。

　　　　2.歸一化

　　　　　　不同行為數據的差別比較大，通過歸一化，數據歸於大致均衡，計算時才能減少異常數據產生的影響。

　　組合不同用戶行為方式：

　　　　1.將不同的行為分組

　　　　2.對不同行為進行加權

四.相似度計算

　　對用戶的行為分析得到用戶的偏好后，可以根據用戶的偏好計算相似用戶和物品，然后可以基於相似用戶或相似物品進行推薦。我們可以將用戶對所有物品的偏好作為一個矩陣來計算用戶之間的相似度，或者將所有用戶對物品的偏好作為一個矩陣來計算物品之間的相似度。

　　1.同現相似度

　　　　指在喜愛物品A的前提下，喜愛物品B的概率。當物品B喜愛率較高時可以使用(A交B)/sqrt(A或B)。

　　2.歐式距離

　　　　1/(1+d(x,y))

　　　　備注：d(x,y) 歐式距離

　　3.皮爾遜相關系數

　　　　皮爾遜相關系數一般用於計算兩個定距變量間聯系的緊密程度，它的取值為【-1~1】之間。

 　　4.Cosine相似度【余弦相似度】

　　　　Cosine相似度廣泛應用於計算文檔數據的相似度。

　　5.Tanimoto系數

　　　　Tanimoto系數也被稱為Jaccard系數，是Cosine相似度的擴展，也多用於計算文檔數據的相似度。

五.代碼實現

package big.data.analyse.ml

import _root_.breeze.numerics.sqrt
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.rdd.RDD

/**
  * 用戶評分
  * @param userid 用戶
  * @param itemid 物品
  * @param pref 評分
  */
case class ItemPref(val userid : String,val itemid : String, val pref : Double) extends Serializable

/**
  * 相似度
  * @param itemid_1 物品
  * @param itemid_2 物品
  * @param similar 相似度
  */
case class ItemSimilar(val itemid_1 : String, val itemid_2 : String, val similar : Double) extends Serializable

/**
  * 給用戶推薦物品
  * @param userid 用戶
  * @param itemid 物品
  * @param pref 推薦系數
  */
case class UserRecommend(val userid : String, val itemid : String, val pref : Double) extends Serializable

/**
  * 相似度計算
  */
class ItemSimilarity extends Serializable{
  def Similarity(user : RDD[ItemPref], stype : String) : RDD[ItemSimilar] = {
    val similar = stype match{
      case "cooccurrence" => ItemSimilarity.CooccurenceSimilarity(user) // 同現相似度
      //case "cosine" => // 余弦相似度
      //case "euclidean" => // 歐式距離相似度
      case _ => ItemSimilarity.CooccurenceSimilarity(user)
    }
    similar
  }
}

object ItemSimilarity{
  def CooccurenceSimilarity(user : RDD[ItemPref]) : (RDD[ItemSimilar]) = {
    val user_1 = user.map(r => (r.userid, r.itemid, r.pref)).map(r => (r._1, r._2))
    /**
      * 內連接，默認根據第一個相同字段為連接條件，物品與物品的組合
      */
    val user_2 = user_1.join(user_1)

    /**
      * 統計
      */
    val user_3 = user_2.map(r => (r._2, 1)).reduceByKey(_+_)

    /**
      * 對角矩陣
      */
    val user_4 = user_3.filter(r => r._1._1 == r._1._2)

    /**
      * 非對角矩陣
      */
    val user_5 = user_3.filter(r => r._1._1 != r._1._2)

    /**
      * 計算相似度
      */
    val user_6 = user_5.map(r => (r._1._1, (r._1._1,r._1._2,r._2)))
      .join(user_4.map(r => (r._1._1, r._2)))

    val user_7 = user_6.map(r => (r._2._1._2, (r._2._1._1, r._2._1._2, r._2._1._3, r._2._2)))
      .join(user_4.map(r => (r._1._1, r._2)))

    val user_8 = user_7.map(r => (r._2._1._1, r._2._1._2, r._2._1._3, r._2._1._4, r._2._2))
      .map(r => (r._1, r._2, (r._3 / sqrt(r._4 * r._5))))

    user_8.map(r => ItemSimilar(r._1, r._2, r._3))
  }
}

class RecommendItem{
  def Recommend(items : RDD[ItemSimilar], users : RDD[ItemPref], number : Int) : RDD[UserRecommend] = {
    val items_1 = items.map(r => (r.itemid_1, r.itemid_2, r.similar))
    val users_1 = users.map(r => (r.userid, r.itemid, r.pref))

    /**
      * i行與j列join
      */
    val items_2 = items_1.map(r => (r._1, (r._2, r._3))).join(users_1.map(r => (r._2, (r._1, r._3))))

    /**
      * i行與j列相乘
      */
    val items_3 = items_2.map(r => ((r._2._2._1, r._2._1._1), r._2._2._2 * r._2._1._2))

    /**
      * 累加求和
      */
    val items_4 = items_3.reduceByKey(_+_)

    /**
      * 過濾已存在的物品
      */
    val items_5 = items_4.leftOuterJoin(users_1.map(r => ((r._1, r._2), 1))).filter(r => r._2._2.isEmpty)
    .map(r => (r._1._1, (r._1._2, r._2._1)))

    /**
      * 分組
      */
    val items_6 = items_5.groupByKey()

    val items_7 = items_6.map(r => {
      val i_2 = r._2.toBuffer
      val i_2_2 = i_2.sortBy(_._2)
      if(i_2_2.length > number){
        i_2_2.remove(0, (i_2_2.length - number))
      }
      (r._1, i_2_2.toIterable)
    })

    val items_8 = items_7.flatMap(r => {
      val i_2 = r._2
      for(v <- i_2) yield (r._1, v._1, v._2)
    })

    items_8.map(r => UserRecommend(r._1, r._2, r._3))
  }
}

/**
  * Created by zhen on 2019/8/9.
  */
object ItemCF {
  def main(args: Array[String]) {
    val conf = new SparkConf()
    conf.setAppName("ItemCF")
    conf.setMaster("local[2]")

    val sc = new SparkContext(conf)

    /**
      * 設置日志級別
      */
    Logger.getRootLogger.setLevel(Level.WARN)

    val array = Array("1,1,0", "1,2,1", "1,4,1", "2,1,0", "2,3,1", "2,4,0", "3,1,0", "3,2,1", "4,1,0", "4,3,1")
    val cf = sc.parallelize(array)

    val user_data = cf.map(_.split(",")).map(r => (ItemPref(r(0), r(1), r(2).toDouble)))

    /**
      * 建立模型
      */
    val mySimilarity = new ItemSimilarity()
    val similarity = mySimilarity.Similarity(user_data, "cooccurrence")

    val recommend = new RecommendItem()
    val recommend_rdd = recommend.Recommend(similarity, user_data, 30)

    /**
      * 打印結果
      */
    println("物品相似度矩陣:" + similarity.count())
    similarity.collect().foreach(record => {
      println(record.itemid_1 +","+ record.itemid_2 +","+ record.similar)
    })

    println("用戶推薦列表:" + recommend_rdd.count())
    recommend_rdd.collect().foreach(record => {
      println(record.userid +","+ record.itemid +","+ record.pref)
    })
  }
}

六.結果