協同過濾算法 R/mapreduce/spark mllib多語言實現

用戶電影評分數據集下載

http://grouplens.org/datasets/movielens/

1) Item-Based,非個性化的，每個人看到的都一樣

2) User-Based，個性化的，每個人看到的不一樣

對用戶的行為分析得到用戶的喜好后，可以根據用戶的喜好計算相似用戶和物品，然后可以基于相似用戶或物品進行推薦。這就是協同過濾中的兩個分支了，基于用戶的和基于物品的協同過濾。

在計算用戶之間的相似度時，是將一個用戶對所有物品的偏好作為一個向量，而在計算物品之間的相似度時，是將所有用戶對某個物品的偏好作為一個向量。求出相似度后，接下來可以求相似鄰居了。

3）基于模型（ModelCF）

        按照模型，可以分為：

        1）最近鄰模型：基于距離的協同過濾算法

       2）Latent Factor Mode（SVD）：基于矩陣分解的模型

       3）Graph：圖模型，社會網絡圖模型

適用場景


     對于一個在線網站，用戶的數量往往超過物品的數量，同時物品數據相對穩定，因此計算物品的相似度不但

     計算量小，同時不必頻繁更新。但是這種情況只適用于電子商務類型的網站，像新聞類，博客等這類網站的

     系統推薦，情況往往是相反的，物品數量是海量的，而且頻繁更新。



r語言實現基于物品的協同過濾算法

    
  

  #引用plyr包
    library(plyr)
    #讀取數據集
    train<-read.table(file="C:/users/Administrator/Desktop/u.data",sep=" ")
    train<-train[1:3]
    
    names(train)<-c("user","item","pref")
    
    #計算用戶列表方法
    usersUnique<-function(){
      users<-unique(train$user)
      users[order(users)]
    }
    
    
    #計算商品列表方法
    itemsUnique<-function(){
      items<-unique(train$item)
      items[order(items)]
    }
    
    # 用戶列表
    users<-usersUnique()
    # 商品列表
    items<-itemsUnique() 
    #建立商品列表索引
    index<-function(x) which(items %in% x)
    data<-ddply(train,.(user,item,pref),summarize,idx=index(item)) 
    
    #同現矩陣
    cooccurrence<-function(data){
      n<-length(items)
      co<-matrix(rep(0,n*n),nrow=n)
      for(u in users){
    idx<-index(data$item[which(data$user==u)])
    m<-merge(idx,idx)
    for(i in 1:nrow(m)){
      co[m$x[i],m$y[i]]=co[m$x[i],m$y[i]]+1
    }
      }
      return(co)
    }
    
    #推薦算法
    recommend<-function(udata=udata,co=coMatrix,num=0){
      n<-length(items)
      
      # all of pref
      pref<-rep(0,n)
      pref[udata$idx]<-udata$pref
      
      # 用戶評分矩陣
      userx<-matrix(pref,nrow=n)
      
      # 同現矩陣*評分矩陣
      r<-co %*% userx
      
      # 推薦結果排序 
      # 把該用戶評分過的商品的推薦值設為0
      r[udata$idx]<-0
      idx<-order(r,decreasing=TRUE)
      topn<-data.frame(user=rep(udata$user[1],length(idx)),item=items[idx],val=r[idx])
      topn<-topn[which(topn$val>0),]
      
      # 推薦結果取前num個
      if(num>0){
    topn<-head(topn,num)
      }
      
      #返回結果
      return(topn)
    }
    
    #生成同現矩陣
    co<-cooccurrence(data) 
    #計算推薦結果
    recommendation<-data.frame()
    for(i in 1:length(users)){
      udata<-data[which(data$user==users[i]),]
      recommendation<-rbind(recommendation,recommend(udata,co,0)) 
    }

mareduce 實現

參考文章：

http://www.cnblogs.com/anny-1980/articles/3519555.html

代碼下載

https://github.com/bsspirit/maven_hadoop_template/releases/tag/recommend

spark ALS實現

Spark mllib里用的是矩陣分解的協同過濾，不是UserBase也不是ItemBase。

參考文章：

http://www.mamicode.com/info-detail-865258.html

import org.apache.spark.SparkConf
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd._
import org.apache.spark.SparkContext
import scala.io.Source


object MovieLensALS {

  def main(args:Array[String]) {

    //設置運行環境

    val sparkConf = new SparkConf().setAppName("MovieLensALS").setMaster("local[5]")

    val sc = new SparkContext(sparkConf)


    //裝載用戶評分，該評分由評分器生成(即生成文件personalRatings.txt)

    val myRatings = loadRatings(args(1))

    val myRatingsRDD = sc.parallelize(myRatings, 1)


    //樣本數據目錄

    val movielensHomeDir = args(0)


    //裝載樣本評分數據，其中最后一列Timestamp取除10的余數作為key，Rating為值，即(Int，Rating)

    val ratings = sc.textFile(movielensHomeDir + "/ratings.dat").map {

      line =>

        val fields = line.split("::")

        // format: (timestamp % 10, Rating(userId, movieId, rating))

        (fields(3).toLong % 10, Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble))

    }


    //裝載電影目錄對照表(電影ID->電影標題)

    val movies = sc.textFile(movielensHomeDir + "/movies.dat").map {

      line =>

        val fields = line.split("::")

        // format: (movieId, movieName)

        (fields(0).toInt, fields(1))

    }.collect().toMap



    //統計有用戶數量和電影數量以及用戶對電影的評分數目

    val numRatings = ratings.count()

    val numUsers = ratings.map(_._2.user).distinct().count()

    val numMovies = ratings.map(_._2.product).distinct().count()

    println("Got " + numRatings + " ratings from " + numUsers + " users " + numMovies + " movies")


    //將樣本評分表以key值切分成3個部分，分別用于訓練 (60%，并加入用戶評分), 校驗 (20%), and 測試 (20%)

    //該數據在計算過程中要多次應用到，所以cache到內存

    val numPartitions = 4

    val training = ratings.filter(x => x._1 < 6).values.union(myRatingsRDD).repartition(numPartitions).persist()

    val validation = ratings.filter(x => x._1 >= 6 && x._1 < 8).values.repartition(numPartitions).persist()

    val test = ratings.filter(x => x._1 >= 8).values.persist()


    val numTraining = training.count()

    val numValidation = validation.count()

    val numTest = test.count()

    println("Training: " + numTraining + " validation: " + numValidation + " test: " + numTest)



    //訓練不同參數下的模型，并在校驗集中驗證，獲取最佳參數下的模型

    val ranks = List(8, 12)

    val lambdas = List(0.1, 10.0)

    val numIters = List(10, 20)

    var bestModel: Option[MatrixFactorizationModel] = None

    var bestValidationRmse = Double.MaxValue

    var bestRank = 0

    var bestLambda = -1.0

    var bestNumIter = -1


    for (rank <- ranks; lambda <- lambdas; numIter <- numIters) {

      val model = ALS.train(training, rank, numIter, lambda)

      val validationRmse = computeRmse(model, validation, numValidation)

      println("RMSE(validation) = " + validationRmse + " for the model trained with rank = "

        + rank + ",lambda = " + lambda + ",and numIter = " + numIter + ".")


      if (validationRmse < bestValidationRmse) {

        bestModel = Some(model)

        bestValidationRmse = validationRmse

        bestRank = rank

        bestLambda = lambda

        bestNumIter = numIter

      }

    }


    //用最佳模型預測測試集的評分，并計算和實際評分之間的均方根誤差（RMSE）

    val testRmse = computeRmse(bestModel.get, test, numTest)

    println("The best model was trained with rank = " + bestRank + " and lambda = " + bestLambda

      + ", and numIter = " + bestNumIter + ", and its RMSE on the test set is " + testRmse + ".")


    //create a naive baseline and compare it with the best model

    val meanRating = training.union(validation).map(_.rating).mean()

    val baselineRmse = math.sqrt(test.map(x => (meanRating - x.rating) * (meanRating - x.rating)).reduce(_ + _) / numTest)

    val improvement = (baselineRmse - testRmse) / baselineRmse * 100

    println("The best model improves the baseline by " + "%1.2f".format(improvement) + "%.")


    //推薦前十部最感興趣的電影，注意要剔除用戶已經評分的電影

    val myRatedMovieIds = myRatings.map(_.product).toSet

    val candidates = sc.parallelize(movies.keys.filter(!myRatedMovieIds.contains(_)).toSeq)

    val recommendations = bestModel.get

      .predict(candidates.map((0, _)))

      .collect()

      .sortBy(-_.rating)

      .take(10)

    var i = 1

    println("Movies recommended for you:")

    recommendations.foreach { r =>

      println("%2d".format(i) + ": " + movies(r.product))

      i += 1

    }


    sc.stop()

  }



  /** 校驗集預測數據和實際數據之間的均方根誤差 **/

  def computeRmse(model:MatrixFactorizationModel,data:RDD[Rating],n:Long):Double = {


    val predictions:RDD[Rating] = model.predict(data.map(x => (x.user,x.product)))

    val predictionsAndRatings = predictions.map{ x =>((x.user,x.product),x.rating)}

      .join(data.map(x => ((x.user,x.product),x.rating))).values

    math.sqrt(predictionsAndRatings.map( x => (x._1 - x._2) * (x._1 - x._2)).reduce(_+_)/n)

  }


  /** 裝載用戶評分文件 personalRatings.txt **/

  def loadRatings(path:String):Seq[Rating] = {

    val lines = Source.fromFile(path).getLines()

    val ratings = lines.map{

      line =>

        val fields = line.split("::")

        Rating(fields(0).toInt,fields(1).toInt,fields(2).toDouble)

    }.filter(_.rating > 0.0)

    if(ratings.isEmpty){

      sys.error("No ratings provided.")

    }else{

      ratings.toSeq

    }

  }

}

參考文章：

http://blog.csdn.net/acdreamers/article/details/44672305

http://www.cnblogs.com/technology/p/4467895.html

http://blog.fens.me/rhadoop-mapreduce-rmr/