代码

  1. import org.apache.spark.ml.recommendation.ALS
  2. import org.apache.spark.SparkConf
  3. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
  4. import org.apache.spark.ml.evaluation.RegressionEvaluator
  6. object ALSDemo {
  7.   case class Rating(userId: Int, movieId: Int, rating: Float, timestamp: Long)
  8.   def main(args: Array[String]): Unit = {
  9.     System.setProperty("hadoop.home.dir", "H:\\winutils\\winutils-master\\hadoop-2.6.0")
  10.     val spark = SparkSession.builder()
  11.       .config(
  12.         new SparkConf()
  13.           .setMaster("local[*]")
  14.           .setAppName("text")
  15.       )
  16.       // 添加满足 inner join 的配置
  17.       .config("spark.sql.crossJoin.enabled","true")
  18.       .getOrCreate()
  19.     import spark.implicits._
  20.     def parseRating(str: String): Rating = {
  21.       val fields = str.split("::")
  22.       Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat, fields(3).toLong)
  23.     }
  25.     val ratings = spark.sparkContext.textFile("data/scala/sample_movielens_ratings.txt")
  26.       .map(parseRating)
  27.       .toDF()
  28.     ratings.show(false)
  29.     val _als: ALS = new ALS()
  30.       // 正则化参数
  31.       .setRegParam(0.01)
  32.       // 最大循环迭代
  33.       .setMaxIter(10)
  34.       // 打分列
  35.       .setRatingCol("rating")
  36.       // 物品列
  37.       .setItemCol("movieId")
  38.       // 用户列
  39.       .setUserCol("userId")
  40.     val Array(tranning, test) = ratings.randomSplit(Array(0.8, 0.2))  // 训练集和测试集
  41.     val model = _als.fit(tranning)  // 训练出模型
  42.     val frame: DataFrame = model.transform(test)  // 预测数据
  43.     frame.show(false)
  44.     // 以方差的方式评估
  45.     val evaluator: RegressionEvaluator = new RegressionEvaluator()
  46.       // 预测列
  47.       .setPredictionCol("prediction")
  48.       // 本来标签列
  49.       .setLabelCol("rating")
  50.       .setMetricName("rmse")
  51.     val d: Double = evaluator.evaluate(frame)
  52.     println(d)
  53.   }
  54. }

ALS补充参数

  1. .setRank() //特征的数量
  2. .setSeed(10) // 种子
  3. .setNonnegative(true) // 求解的最小二乘值是否非负
  4. .setRegParam(0.01) // 正则化
  5. .setMaxIter(10) // 迭代次数
  6. .setRatingCol(“rating”) // 打分列
  7. .setItemCol(“movieId”) // 物品列
  8. .setUserCol(“userId”) // 用户列

  9. setImplicitPrefs() // 显示反馈还是隐示反馈

    ALS特殊补充(源码方法)

    1. @Since("1.3.0")
    2. override def transformSchema(schema: StructType): StructType = {
    3.  // user and item will be cast to Int
    4.  // user 和 item 必须是 Int类型不能是string
    5.  SchemaUtils.checkNumericType(schema, $(userCol))
    6.  SchemaUtils.checkNumericType(schema, $(itemCol))
    7.  // 返回的预测列是float类型
    8.  SchemaUtils.appendColumn(schema, $(predictionCol), FloatType)
    9. }

  10. transformSchema 检查userID， itemID 是否为Int

  11. ALS.train 模型训练 得到用户特征矩阵和物品特征矩阵
  12. new ALSModel 创建ALS模型

    ALS 默认参数

    1. ratings: RDD[Rating[ID]],
    2. rank: Int = 10,
    3. numUserBlocks: Int = 10,
    4. numItemBlocks: Int = 10,
    5. maxIter: Int = 10,
    6. regParam: Double = 1.0,
    7. implicitPrefs: Boolean = false,
    8. alpha: Double = 1.0,
    9. nonnegative: Boolean = false,
    10. intermediateRDDStorageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK,
    11. finalRDDStorageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK,
    12. checkpointInterval: Int = 10,