利用ALS算法做用戶產品推送

ALS 是什么？ 
ALS 是交替最小二乘 （alternating least squares）的簡稱。 
在機器學習的上下文中，ALS 特指使用交替最小二乘求解的一個協同推薦算法。 
它通過觀察到的所有用戶給產品的打分，來推斷每個用戶的喜好並向用戶推薦適合的產品。

協同過濾 常被應用於推薦系統，旨在補充用戶-商品關聯矩陣中所缺失的部分。 
MLlib當前支持基於模型的協同過濾，其中用戶和商品通過一小組隱語義因子進行表達， 
並且這些因子也用於預測缺失的元素。Spark MLlib實現了 交替最小二乘法 (ALS) 
來學習這些隱性語義因子。在 MLlib 中的實現有如下的參數: 
•numBlocks 是用於並行化計算的分塊個數 (設置為-1，為自動配置)。 
•rank 是模型中隱語義因子的個數。 
•iterations 是迭代的次數。 
•lambda 是ALS的正則化參數。 
•implicitPrefs 決定了是用顯性反饋ALS的版本還是用適用隱性反饋數據集的版本。 
•alpha 是一個針對於隱性反饋 ALS 版本的參數，這個參數決定了偏好行為強度的基准。

可以調整這些參數，不斷優化結果，使均方差變小。比如：iterations越多，lambda較小，均方差會較小，推薦結果較優。

注意上面說可以利用隱性因子預測缺失元素。下面結合數據講解下： 
1,1,5.0 
1,2,1.0 
1,3,5.0 
1,4,1.0 
2,2,1.0 
2,3,5.0 
2,4,1.0 
3,1,1.0 
3,2,5.0 
3,3,1.0 
3,4,5.0 
4,1,1.0 
4,2,5.0 
4,3,1.0 
4,4,5.0 
第一個數據，表示用戶id，第二數據則表示，產品id，第三則是評分。 
注意這里缺失第二個用戶對第一個產品的評分，算法仍然可以根據第二用戶與其他用戶的評分相似度，從而預測對第一個產品的評分，從而根據評分是否要將產品進行推送。

import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.mllib.recommendation.{ALS, Rating}
/**
 * Created by Administrator on 2016/8/16.
 */
object RecommendationExample2 {
  def main(args: Array[String]) {
    val conf =new SparkConf().setAppName("RecommendationExample2").setMaster("local");
    val sc = new SparkContext(conf)

    //加載數據
    val data = sc.textFile("file///F:/1/alstest.txt")

    val ratings=data.map(_.split(",")match{case Array(user,item,rate)=>
      Rating(user.toInt,item.toInt,rate.toDouble)
    })

    // 訓練模型
    val rank = 10//是模型中隱語義因子的個數。
    val numIterations = 5
    val lambda =0.01  //是ALS的正則化參數。

    val model=ALS.train(ratings,rank,numIterations,lambda)

    val usersProducts=ratings.map{case Rating(user,item,rate)=>
      (user,item)
    }

    val predictions =model.predict(usersProducts)
   // println("_----------------------------------predictions-------------------------------")
    //predictions.foreach(println)

    //這里的結果其實和上面的結果沒有太多的不同，只是將評分與用戶id產品id分割了下
    val predictions2=model.predict(usersProducts).map{case Rating(user,item,rate)=>
      ((user,item),rate)
    }
    //println("------------------------------predictions2-------------------------------")
   // predictions2.foreach(println)

    //合並原始數據和評分
    val ratesAndPreds=ratings.map{case Rating(user,item,rate)=>
      ((user,item),rate)
    }.join(predictions2)//注意要是同類的才能合並，這里就不能直接合並predictions

   // println("=========================ratesAndPreds=========================")
   // ratesAndPreds foreach(println)

    //然后計算均方差，注意這里沒有調用 math.sqrt方法
    val MSE=ratesAndPreds.map{case((user,item),(r1,r2))=>
      val err=r1-r2
        err*err
    }.mean()
  //打印出方差結果
    println("Mean Squared Error = " + MSE)

    //獲得用戶的id
  val users=data.map(_.split(",")match {
    case Array(user,item,rate)=>(user)
  }).distinct().collect()

    //循環用戶，將id傳給模型
    users.foreach(
    user=>{
      var rs=model.recommendProducts(user.toInt,numIterations)//通過模型對用戶進行商品的評分推送
      var value = ""
      var key = 0

      rs.foreach(
        r=>{
          key=r.user
          value=value+r.product+":"+r.rating+","
        })
      println(key.toString+"   " + value)

    })

    /*val rs=model.recommendProductsForUsers(numIterations)
    rs.foreach(r=>

      r._2.foreach(x=>
        print(x.toString)
      )

      //println(r._1+":"+r._2.toString)
    )*/
  }
}

 

轉自：http://blog.csdn.net/young_so_nice/article/details/52221324