首先給出程式碼：

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.ml.recommendation._
import org.apache.spark.mllib.recommendation.{Rating,ALS,MatrixFactorizationModel}
import org.apache.spark.rdd._
import org.apache.spark.mllib.recommendation.Rating
/***
 *   time   : 2016年5月4日
 *  author  : hongdd
 * function : spark使用者推薦系統
 */
 

object FilmAdv{
  def main(args: Array[String]): Unit = {
    //執行環境
    val sconf = new SparkConf().setMaster("local[4]").setAppName("MLlib")
    val sc = new SparkContext(sconf)

    //載入使用者評分資料
    val data= sc.textFile("D:\\movie\\ratings.dat")
    val ratings = data.map(_.split("::") match {
      case
 
 Array(user,item,rate,time) => Rating(user.toInt,item.toInt,rate.toDouble)
    }).cache();
    //統計人數、商品數和資料總量
    val users=ratings.map(_.user).distinct().count();
    val products=ratings.map(_.product).distinct().count();
    val rates = ratings.count();
    println("count:"+rates+",from user:"+users+" with pro:"
 
+products)

    //隨機抽取訓練集和測試集
    //randomSplit 第一個引數是分組每份的許可權,和必須為1;第二個引數可省略,是隨機數種子
    val splits=ratings.randomSplit(Array(0.8,0.2), seed=111L)
    val numPartitions=4
    val training=splits(0).repartition(numPartitions)
    val test=splits(1).repartition(numPartitions)

    //訓練模型
    val rank=12  //隱藏因子數
    val lambad=0.01  //ALS演算法中正則化引數
    val numIteration=5 //迭代次數    越大消耗記憶體越多
    val model=ALS.train(ratings, rank, numIteration,lambad)
    //使用者和商品的特徵向量
    //model.userFeatures
    //model.productFeatures

    //評分初始資料中的(使用者,商品)列表
    val userProducts=ratings.map{case Rating(user,product,rate) => (user,product)}

    //使用模型對使用者商品就行預測
    var predictions=model.predict(userProducts).map { case Rating(user,product,rate) => ((user,product),rate) }

    //合併真實評分和預測評分資料
    val ratesAndPreds=ratings.map { case Rating(user,product,rate) => ((user,product),rate) }.join(predictions)

    //模型指標
    val rmse = math.sqrt(ratesAndPreds.map{case ((user,product),(r1,r2)) => 
      val err=(r1-r2)
      err*err
    }.mean())

    println(s"RMSE=$rmse")

    //儲存評分結果
//    ratesAndPreds.sortByKey().repartition(1).sortBy(_._1).map({
//      case ((user,product),(rate,pred)) => (user+","+product+","+rate+","+pred) 
//    }).saveAsTextFile("")

    //位使用者推薦商品
    val topK = model.recommendProducts(1, 3);
    topK.foreach { println }

  }
}

演算法核心思想就是：通過你已經評論過得電影，找到和你有相同喜好的其他人，通過他們的評分資料來預測你對還未看過的評分，將分數從高往低排序就是給你推薦的電影順序。