美文网首页
基于Spark的电影推荐系统(推荐系统~7)

基于Spark的电影推荐系统(推荐系统~7)

作者: 留歌_36 | 来源:发表于2019-10-21 13:53 被阅读0次

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

    • <font color=#00cc66 size=4 face="黑体"> 本模块基于第4节得到的模型,开始为用户做实时推荐,推荐用户最有可能喜爱的5部电影。</font>

    说明几点

    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()
        }
    }
    
    

    步骤三:将创建的项目进行打包上传到服务器
    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

    相关文章

      网友评论

          本文标题:基于Spark的电影推荐系统(推荐系统~7)

          本文链接:https://www.haomeiwen.com/subject/lortvctx.html