基于Spark的电影推荐系统(推荐系统~7)
22/100
发布文章
liuge36

第四部分-推荐系统-实时推荐

  • 本模块基于第4节得到的模型,开始为用户做实时推荐,推荐用户最有可能喜爱的5部电影。

    说明几点

    1.数据来源是 testData 测试集的数据。这里面的用户,可能存在于训练集中,也可能是新用户。因此,这里要做处理。
  1. 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()
    }
}

步骤三:将创建的项目进行打包上传到服务器

步骤四:先编写个性化推荐代码 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 测试集的数据。这里面的用户,可能存在于训练集中,也可能是新用户。因此,这里要做处理。

  1. 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.createDirectStreamString, String, StringDecoder, StringDecoder

    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 段落

01-14 01:05