基於Spark的電影推薦系統（推薦系統~7）

基於Spark的電影推薦系統（推薦系統~7）
22/100
發布文章
liuge36

第四部分-推薦系統-實時推薦

• 本模塊基於第4節得到的模型，開始為用戶做實時推薦，推薦用戶最有可能喜愛的5部電影。

說明幾點

1.數據來源是 testData 測試集的數據。這里面的用戶，可能存在於訓練集中，也可能是新用戶。因此，這里要做處理。
2. SparkStreaming + kakfa

開始Coding

步驟一：在streaming 包下，新建PopularMovies2

package com.csylh.recommend.streaming

import com.csylh.recommend.config.AppConf
import org.apache.spark.sql.SaveMode

/**
  * Description: 個性化推薦
  *
  * @Author: 留歌36
  * @Date: 2019/10/18 17:42
  */
object PopularMovies2 extends AppConf{
    def main(args: Array[String]): Unit = {
        val movieRatingCount = spark.sql("select count(*) c, movieid from trainingdata group by movieid order by c")
        // 前5部進行推薦
        val Top5Movies = movieRatingCount.limit(5)

        Top5Movies.registerTempTable("top5")

        val top5DF = spark.sql("select a.title from movies a join top5 b on a.movieid=b.movieid")

        // 把數據寫入到HDFS上
        top5DF.write.mode(SaveMode.Overwrite).parquet("/tmp/top5DF")

        // 將數據從HDFS加載到Hive數據倉庫中去
        spark.sql("drop table if exists top5DF")
        spark.sql("create table if not exists top5DF(title string) stored as parquet")
        spark.sql("load data inpath '/tmp/top5DF' overwrite into table top5DF")

        // 最終表里應該是5部推薦電影的名稱
    }
}

步驟二：在streaming 包下，新建SparkDirectStreamApp

package com.csylh.recommend.streaming

import com.csylh.recommend.config.AppConf
import kafka.serializer.StringDecoder
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Description:
  *
  * @Author: 留歌36
  * @Date: 2019/10/18 16:33
  */
object SparkDirectStreamApp extends AppConf{
    def main(args:Array[String]): Unit ={
      val ssc = new StreamingContext(sc, Seconds(5))

      val topics = "movie_topic".split(",").toSet

      val kafkaParams = Map[String, String](
        "metadata.broker.list"->"hadoop001:9093,hadoop001:9094,hadoop001:9095",
        "auto.offset.reset" -> "largest" //smallest :從頭開始 largest：最新
      )
      // Direct 模式：SparkStreaming 主動去Kafka中pull拉數據
      val modelPath = "/tmp/BestModel/0.8521581387523667"
      val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

      def exist(u: Int): Boolean = {
        val trainingdataUserIdList = spark.sql("select distinct(userid) from trainingdata")
          .rdd
          .map(x => x.getInt(0))
          .collect()  // RDD[row] ==> RDD[Int]

        trainingdataUserIdList.contains(u)
      }

      // 為沒有登錄的用戶推薦電影的策略：
      // 1.推薦觀看人數較多的電影，采用這種策略
      // 2.推薦最新的電影
      val defaultrecresult = spark.sql("select * from top5DF").rdd.toLocalIterator

      // 創建SparkStreaming接收kafka消息隊列數據的2種方式
      // 一種是Direct approache,通過SparkStreaming自己主動去Kafka消息隊
      // 列中查詢還沒有接收進來的數據，並把他們拉pull到sparkstreaming中。

      val model = MatrixFactorizationModel.load(ssc.sparkContext, modelPath)
      val messages = stream.foreachRDD(rdd=> {

              val userIdStreamRdd = rdd.map(_._2.split("|")).map(x=>x(1)).map(_.toInt)

              val validusers = userIdStreamRdd.filter(userId => exist(userId))
              val newusers = userIdStreamRdd.filter(userId => !exist(userId))

              // 采用迭代器的方式來避開對象不能序列化的問題。
              // 通過對RDD中的每個元素實時產生推薦結果，將結果寫入到redis，或者其他高速緩存中，來達到一定的實時性。
              // 2個流的處理分成2個sparkstreaming的應用來處理。
              val validusersIter = validusers.toLocalIterator
              val newusersIter = newusers.toLocalIterator

              while (validusersIter.hasNext) {
                val u= validusersIter.next
                println("userId"+u)
                val recresult = model.recommendProducts(u, 5)
                val recmoviesid = recresult.map(_.product)
                println("我為用戶" + u + "【實時】推薦了以下5部電影：")
                for (i <- recmoviesid) {
                  val moviename = spark.sql(s"select title from movies where movieId=$i").first().getString(0)
                  println(moviename)
                }
              }

              while (newusersIter.hasNext) {
                println("*新用戶你好*以下電影為您推薦below movies are recommended for you :")
                for (i <- defaultrecresult) {
                  println(i.getString(0))
                }
              }


     })
      ssc.start()
      ssc.awaitTermination()
    }
}

步驟三：將創建的項目進行打包上傳到服務器
mvn clean package -Dmaven.test.skip=true

步驟四：先編寫個性化推薦代碼 shell 執行腳本

[root@hadoop001 ml]# vim PopularMovies2.sh 
export HADOOP_CONF_DIR=/root/app/hadoop-2.6.0-cdh5.7.0/etc/hadoop

$SPARK_HOME/bin/spark-submit \
--class com.csylh.recommend.streaming.PopularMovies2 \
--master spark://hadoop001:7077 \
--name PopularMovies2 \
--driver-memory 10g \
--executor-memory 5g \
/root/data/ml/movie-recommend-1.0.jar

步驟五：執行sh PopularMovies2.sh

確保：

[root@hadoop001 ml]# spark-sql
19/10/20 22:59:28 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark master: local[*], Application Id: local-1571583574311
spark-sql> show tables;
default	links	false
default	movies	false
default	ratings	false
default	tags	false
default	testdata	false
default	top5df	false
default	trainingdata	false
default	trainingdataasc	false
default	trainingdatadesc	false
Time taken: 2.232 seconds, Fetched 9 row(s)
spark-sql> select * from top5df;
Follow the Bitch (1996)
Radio Inside (1994)
Faces of Schlock (2005)
Mág (1988)
"Son of Monte Cristo
Time taken: 1.8 seconds, Fetched 5 row(s)
spark-sql> 

步驟六：再編寫model實時推薦代碼 shell 執行腳本

export HADOOP_CONF_DIR=/root/app/hadoop-2.6.0-cdh5.7.0/etc/hadoop

$SPARK_HOME/bin/spark-submit \
--class com.csylh.recommend.streaming.SparkDirectStreamApp \
--master spark://hadoop001:7077 \
--name SparkDirectStreamApp \
--driver-memory 10g \
--executor-memory 5g \
--total-executor-cores 10 \
--jars /root/app/kafka_2.11-1.1.1/libs/kafka-clients-1.1.1.jar \
--packages "mysql:mysql-connector-java:5.1.38,org.apache.spark:spark-streaming-kafka-0-8_2.11:2.4.2" \
/root/data/ml/movie-recommend-1.0.jar

步驟七：sh SparkDirectStreamApp.sh

// TODO...

有任何問題，歡迎留言一起交流~~
更多文章：基於Spark的電影推薦系統：https://blog.csdn.net/liuge36/column/info/29285

第四部分-推薦系統-實時推薦
本模塊基於第4節得到的模型，開始為用戶做實時推薦，推薦用戶最有可能喜愛的5部電影。
說明幾點
1.數據來源是 testData 測試集的數據。這里面的用戶，可能存在於訓練集中，也可能是新用戶。因此，這里要做處理。
2. SparkStreaming + kakfa
在這里插入圖片描述

開始Coding
步驟一：在streaming 包下，新建PopularMovies2

package com.csylh.recommend.streaming

import com.csylh.recommend.config.AppConf
import org.apache.spark.sql.SaveMode

/**

• Description: 個性化推薦

• @Author: 留歌36

• @Date: 2019/10/18 17:42
  /
  object PopularMovies2 extends AppConf{
  def main(args: Array[String]): Unit = {
  val movieRatingCount = spark.sql("select count() c, movieid from trainingdata group by movieid order by c")
  // 前5部進行推薦
  val Top5Movies = movieRatingCount.limit(5)

  Top5Movies.registerTempTable("top5")
  
  val top5DF = spark.sql("select a.title from movies a join top5 b on a.movieid=b.movieid")
  
  // 把數據寫入到HDFS上
  top5DF.write.mode(SaveMode.Overwrite).parquet("/tmp/top5DF")
  
  // 將數據從HDFS加載到Hive數據倉庫中去
  spark.sql("drop table if exists top5DF")
  spark.sql("create table if not exists top5DF(title string) stored as parquet")
  spark.sql("load data inpath '/tmp/top5DF' overwrite into table top5DF")
  
  // 最終表里應該是5部推薦電影的名稱
  

  }
  }

步驟二：在streaming 包下，新建SparkDirectStreamApp

package com.csylh.recommend.streaming

import com.csylh.recommend.config.AppConf
import kafka.serializer.StringDecoder
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**

• Description:

• @Author: 留歌36

• @Date: 2019/10/18 16:33
  */
  object SparkDirectStreamApp extends AppConf{
  def main(args:Array[String]): Unit ={
  val ssc = new StreamingContext(sc, Seconds(5))

  val topics = "movie_topic".split(",").toSet

  val kafkaParams = Map[String, String](
  "metadata.broker.list"->"hadoop001:9093,hadoop001:9094,hadoop001:9095",
  "auto.offset.reset" -> "largest" //smallest :從頭開始 largest：最新
  )
  // Direct 模式：SparkStreaming 主動去Kafka中pull拉數據
  val modelPath = "/tmp/BestModel/0.8521581387523667"
  val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

  def exist(u: Int): Boolean = {
  val trainingdataUserIdList = spark.sql("select distinct(userid) from trainingdata")
  .rdd
  .map(x => x.getInt(0))
  .collect() // RDD[row] ==> RDD[Int]

  trainingdataUserIdList.contains(u)
  

  }

  // 為沒有登錄的用戶推薦電影的策略：
  // 1.推薦觀看人數較多的電影，采用這種策略
  // 2.推薦最新的電影
  val defaultrecresult = spark.sql("select * from top5DF").rdd.toLocalIterator

  // 創建SparkStreaming接收kafka消息隊列數據的2種方式
  // 一種是Direct approache,通過SparkStreaming自己主動去Kafka消息隊
  // 列中查詢還沒有接收進來的數據，並把他們拉pull到sparkstreaming中。

  val model = MatrixFactorizationModel.load(ssc.sparkContext, modelPath)
  val messages = stream.foreachRDD(rdd=> {

        val userIdStreamRdd = rdd.map(_._2.split("|")).map(x=>x(1)).map(_.toInt)
  
        val validusers = userIdStreamRdd.filter(userId => exist(userId))
        val newusers = userIdStreamRdd.filter(userId => !exist(userId))
  
        // 采用迭代器的方式來避開對象不能序列化的問題。
        // 通過對RDD中的每個元素實時產生推薦結果，將結果寫入到redis，或者其他高速緩存中，來達到一定的實時性。
        // 2個流的處理分成2個sparkstreaming的應用來處理。
        val validusersIter = validusers.toLocalIterator
        val newusersIter = newusers.toLocalIterator
  
        while (validusersIter.hasNext) {
          val u= validusersIter.next
          println("userId"+u)
          val recresult = model.recommendProducts(u, 5)
          val recmoviesid = recresult.map(_.product)
          println("我為用戶" + u + "【實時】推薦了以下5部電影：")
          for (i <- recmoviesid) {
            val moviename = spark.sql(s"select title from movies where movieId=$i").first().getString(0)
            println(moviename)
          }
        }
  
        while (newusersIter.hasNext) {
          println("*新用戶你好*以下電影為您推薦below movies are recommended for you :")
          for (i <- defaultrecresult) {
            println(i.getString(0))
          }
        }
  

  })
  ssc.start()
  ssc.awaitTermination()
  }
  }

步驟三：將創建的項目進行打包上傳到服務器
mvn clean package -Dmaven.test.skip=true

步驟四：先編寫個性化推薦代碼 shell 執行腳本

[root@hadoop001 ml]# vim PopularMovies2.sh
export HADOOP_CONF_DIR=/root/app/hadoop-2.6.0-cdh5.7.0/etc/hadoop

$SPARK_HOME/bin/spark-submit
--class com.csylh.recommend.streaming.PopularMovies2
--master spark://hadoop001:7077
--name PopularMovies2
--driver-memory 10g
--executor-memory 5g
/root/data/ml/movie-recommend-1.0.jar

步驟五：執行sh PopularMovies2.sh

確保：

[root@hadoop001 ml]# spark-sql
19/10/20 22:59:28 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark master: local[*], Application Id: local-1571583574311
spark-sql> show tables;
default links false
default movies false
default ratings false
default tags false
default testdata false
default top5df false
default trainingdata false
default trainingdataasc false
default trainingdatadesc false
Time taken: 2.232 seconds, Fetched 9 row(s)
spark-sql> select * from top5df;
Follow the Bitch (1996)
Radio Inside (1994)
Faces of Schlock (2005)
Mág (1988)
"Son of Monte Cristo
Time taken: 1.8 seconds, Fetched 5 row(s)
spark-sql>

步驟六：再編寫model實時推薦代碼 shell 執行腳本

export HADOOP_CONF_DIR=/root/app/hadoop-2.6.0-cdh5.7.0/etc/hadoop

$SPARK_HOME/bin/spark-submit
--class com.csylh.recommend.streaming.SparkDirectStreamApp
--master spark://hadoop001:7077
--name SparkDirectStreamApp
--driver-memory 10g
--executor-memory 5g
--total-executor-cores 10
--jars /root/app/kafka_2.11-1.1.1/libs/kafka-clients-1.1.1.jar
--packages "mysql:mysql-connector-java:5.1.38,org.apache.spark:spark-streaming-kafka-0-8_2.11:2.4.2"
/root/data/ml/movie-recommend-1.0.jar
步驟七：sh SparkDirectStreamApp.sh

// TODO…

有任何問題，歡迎留言一起交流~~
更多文章：基於Spark的電影推薦系統：https://blog.csdn.net/liuge36/column/info/29285

Markdown 5232 字數 214 行數 當前行 1, 當前列 0HTML 4908 字數 154 段落
譯