数据

data.txt

思路

  1. 加载数据
  2. 获取用户-电影-评分
  3. 获取电影-电影相似度

    加载数据

    ```java val spark = SparkSession.builder() .config( new SparkConf()
    1.  .setMaster("local[*]")
    2.  .setAppName("text")
    ) // 添加满足 inner join 的配置 .config(“spark.sql.crossJoin.enabled”,”true”) .getOrCreate() import spark.implicits._
  1. def parseRating(str: String): Rating = {
  2.   val fields = str.split("::")
  3.   Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat, fields(3).toLong)
  4. }
  6. val ratings = spark.sparkContext.textFile("data/scala/sample_movielens_ratings.txt")
  7.   .map(parseRating)
  8.   .toDF()
  9. ratings.show(false)
  1. <a name="QNXXp"></a>
  2. # 第二步求得 movieId_1 和 movieId_2 的共同个数
  3. ```java
  4.  // 物品相似度表(矩阵)
  5.     val dataFrame: DataFrame = ratings.select("userId", "movieId")    
  6.     val frame: DataFrame = dataFrame
  7.       .join(dataFrame, "userId")
  8.       .map(x => {
  9.         (x(0).toString, (x(1).toString, x(2).toString))
  10.       })
  11.       .map(x => {
  12.         (x._2._1, x._2._2, 1)
  13.       })
  14.       .groupBy("_1", "_2").agg(sum("_3").as("sim"))
  15.       .withColumnRenamed("_1", "movieId_1")
  16.       .withColumnRenamed("_2", "movieId_2")

部分结果

image.png

第三步 得出结果

  1. 形成 userId movieId rating movieId_1 movieId_2 sim
  2. userId | movieId_2 | interse interse = rating * sim
  3. 利用开窗函数获得前四
    1.  ratings.join(frame, ratings("movieId") === frame("movieId_1"), "inner" )
    2.    // 获取sim与rating乘积
    3.    .withColumn("interse", frame("sim")*ratings("rating"))
    4.    .select("userId", "movieId_2", "interse")
    5.    // 按照用户与movieId_2  分组 求 interse得总和 并且获取top4
    6.    .groupBy("userId", "movieId_2")
    7.    .agg(sum("interse").as("interse"))
    8.    // 开窗函数
    9.    .withColumn("top",  row_number().over(Window.partitionBy("userId").orderBy($"interse".desc)))
    10.    .where($"top" < 5)
    11.    .show(false)
    12.  /*如果推荐的前四电影中 用户没有看过则 这个电影就是推荐给用户得其中之一*/

    一部分结果

    image.png

得出结果中间一部分代码得图片

  1. ratings.join(frame, ratings("movieId") === frame("movieId_1"), "inner" )
  2.   // 获取sim与rating乘积
  3.   .withColumn("interse", frame("sim")*ratings("rating")).show(false)

image.png
之后对这个表进行操作得到推荐表 我写在了一起

整体代码

  1. import org.apache.spark.SparkConf
  2. import org.apache.spark.ml.recommendation.ALS
  3. import org.apache.spark.sql.expressions.Window
  4. import org.apache.spark.sql.functions._
  5. import org.apache.spark.sql.{DataFrame, SparkSession}
  7. object hello {
  8.   case class Rating(userId: Int, movieId: Int, rating: Float, timestamp: Long)
  10.   def main(args: Array[String]): Unit = {
  11.     // 机器学习 基于协同过滤推荐系统
  12.     System.setProperty("hadoop.home.dir", "H:\\winutils\\winutils-master\\hadoop-2.6.0")
  13.     val spark = SparkSession.builder()
  14.       .config(
  15.         new SparkConf()
  16.           .setMaster("local[*]")
  17.           .setAppName("text")
  18.       )
  19.       // 添加满足 inner join 的配置
  20.       .config("spark.sql.crossJoin.enabled", "true")
  21.       .getOrCreate()
  22.     import spark.implicits._
  25.     def parseRating(str: String): Rating = {
  26.       val fields = str.split("::")
  27.       Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat, fields(3).toLong)
  28.     }
  30.     var ratings = spark.sparkContext.textFile("data/scala/sample_movielens_ratings.txt")
  31.       .map(parseRating)
  32.       .toDF()
  33.     ratings = ratings.select("userId","movieId","rating").where($"rating" > 0)
  35. //    val useItemDate = ratings.groupBy("userId")
  36. //      .pivot("movieId")
  37. //      .sum("rating")
  38. //      .na
  39. //      .fill(0)
  40. //    useItemDate.show(false)
  42.     // 物品相似度表(矩阵)
  43.     val dataFrame: DataFrame = ratings.select("userId", "movieId")
  45.     val frame: DataFrame = dataFrame
  46.       .join(dataFrame, "userId")
  47.       .map(x => {
  48.         (x(0).toString, (x(1).toString, x(2).toString))
  49.       })
  50.       .map(x => {
  51.         (x._2._1, x._2._2, 1)
  52.       })
  53.       .groupBy("_1", "_2").agg(sum("_3").as("sim"))
  54.       .withColumnRenamed("_1", "movieId_1")
  55.       .withColumnRenamed("_2", "movieId_2")
  56.     frame.show(false)
  60.     ratings.join(frame, ratings("movieId") === frame("movieId_1"), "inner" )
  61.       // 获取sim与rating乘积
  62.       .withColumn("interse", frame("sim")*ratings("rating"))
  63.       .select("userId", "movieId_2", "interse")
  64.       // 按照用户与movieId_2  分组 求 interse得总和 并且获取top4
  65.       .groupBy("userId", "movieId_2")
  66.       .agg(sum("interse").as("interse"))
  67.       // 开窗函数
  68.       .withColumn("top",  row_number().over(Window.partitionBy("userId").orderBy($"interse".desc)))
  69.       .where($"top" < 5)
  70.       .show(false)
  71.     /*如果推荐的前四电影中 用户没有看过则 这个电影就是推荐给用户得其中之一*/
  73.     ratings.join(frame, ratings("movieId") === frame("movieId_1"), "inner" )
  74.       // 获取sim与rating乘积
  75.       .withColumn("interse", frame("sim")*ratings("rating")).show(false)
  78. //    val frame1: DataFrame = frame.groupBy("movieId_1")
  79. //      .pivot("movieId_2")
  80. //      .sum("sim")
  81. //      .withColumnRenamed("movieId_1", "movieId")
  82. //      .na
  83. //      .fill(0)
  84. //    frame1.show(false)
  86.   }
  88. }

注意

  1. 一定要注意表的对应列和对应关系
  2. 联表注意联表的两列不是我们需要的 而是附带的另一列 比如联表是(movieId, movieId_1) 我们需要的是 (movieId_2)