使用spark mllib中协同过滤推荐算法ALS建立推荐模型

package com.yyds.tags.ml.rs.rdd

import org.apache.spark.mllib.evaluation.RegressionMetrics
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


object SparkAlsRmdAndMovie {

  def main(args: Array[String]): Unit = {

    // 1 构建SparkContext对象
    val sc: SparkContext = {
      // 创建spark conf对象
      val sparkConf: SparkConf = new SparkConf()
        .setMaster("local[4]")
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))

      val context: SparkContext = SparkContext.getOrCreate(sparkConf)
      // 设置检查点
      context.setCheckpointDir("data/ckpt/als-ml-" + System.currentTimeMillis())
      context
    }

    // 读取电影评分数据
    val rawRDD: RDD[String] = sc.textFile("datas/u.data")

    println(s"Count = ${rawRDD.count()}")
    println(s"First = ${rawRDD.first()}")


    // 数据转换，构建RDD[Rating]
    val ratingRDD: RDD[Rating] = rawRDD
          //过滤
          .filter(line => null != line && line.split("\t").length == 4)
        .map{
          line =>
            val Array(userId,movieId,rating,_) = line.split("\t")
            Rating(userId.toInt,movieId.toInt,rating.toDouble)
        }

    // 划分训练集和测试集
    val Array(trainRatings,testRatings) = ratingRDD.randomSplit(Array(0.8, 0.2))

    val alsModel: MatrixFactorizationModel = ALS.train(
      ratings = trainRatings,
      rank = 10, // 特征数
      iterations = 20 // 迭代次数
    )

    // 获取因子矩阵
    val userFeatures: RDD[(Int, Array[Double])] = alsModel.userFeatures
    // 获取物品因子矩阵
    val productFeatures: RDD[(Int, Array[Double])] = alsModel.productFeatures

    // TODO 模型评估
    val realRatingsRDD: RDD[((Int, Int), Double)] = testRatings.map {
      tuple =>
        ((tuple.user, tuple.product), tuple.rating)
    }

    // 使用模型对测试数据进行预测
    val predictRatingsRDD: RDD[((Int, Int), Double)] = alsModel
      .predict(realRatingsRDD.map(_._1))
      .map(tuple =>
        ((tuple.user, tuple.product), tuple.rating)
      )
    // 合并预测值和真实值
    val predictAndRealRDD: RDD[((Int, Int), (Double, Double))] = predictRatingsRDD
      .join(realRatingsRDD)
    val metrics: RegressionMetrics = new RegressionMetrics(predictAndRealRDD.map(_._2))
    println(s"RMSE = ${metrics.rootMeanSquaredError}")


    println("----------------预测196用户对电影242的评分-----------------------")

    // TODO 预测
    val predict: Double = alsModel.predict(196, 242)
    println(s"预测196用户对电影242的评分： $predict")

    println("----------------为196用户推荐的10部电影-----------------------")

    // 为196用户推荐的10部电影
    val movies: Array[Rating] = alsModel.recommendProducts(196, 10)
    movies.foreach(println)

    println("----------------为242电影推荐的10个用户-----------------------")

    // 为242电影推荐的10个用户
    val users: Array[Rating] = alsModel.recommendUsers(242, 10)
    users.foreach(println)

    // TODO 保存和加载模型
    val path = "datas/als/ml-als-model-" + System.currentTimeMillis()
    alsModel.save(sc,path)

    // 加载
//    val model: MatrixFactorizationModel = MatrixFactorizationModel.load(sc, path)



    // 休眠
    Thread.sleep(10000000L)
    sc.stop()
  }

}