第四部分-推薦系統(tǒng)-實(shí)時(shí)推薦
- <font color=#00cc66 size=4 face="黑體"> 本模塊基于第4節(jié)得到的模型，開(kāi)始為用戶做實(shí)時(shí)推薦茂洒，推薦用戶最有可能喜愛(ài)的5部電影威根。</font>
說(shuō)明幾點(diǎn)
1.數(shù)據(jù)來(lái)源是 testData 測(cè)試集的數(shù)據(jù)。這里面的用戶嘴脾，可能存在于訓(xùn)練集中男摧，也可能是新用戶。因此译打，這里要做處理耗拓。
-
SparkStreaming + kakfa
在這里插入圖片描述
開(kāi)始Coding
步驟一：在streaming 包下，新建PopularMovies2
package com.csylh.recommend.streaming
import com.csylh.recommend.config.AppConf
import org.apache.spark.sql.SaveMode
/**
* Description: 個(gè)性化推薦
*
* @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部進(jìn)行推薦
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")
// 把數(shù)據(jù)寫(xiě)入到HDFS上
top5DF.write.mode(SaveMode.Overwrite).parquet("/tmp/top5DF")
// 將數(shù)據(jù)從HDFS加載到Hive數(shù)據(jù)倉(cāng)庫(kù)中去
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")
// 最終表里應(yīng)該是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 :從頭開(kāi)始 largest：最新
)
// Direct 模式：SparkStreaming 主動(dòng)去Kafka中pull拉數(shù)據(jù)
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)
}
// 為沒(méi)有登錄的用戶推薦電影的策略：
// 1.推薦觀看人數(shù)較多的電影乔询，采用這種策略
// 2.推薦最新的電影
val defaultrecresult = spark.sql("select * from top5DF").rdd.toLocalIterator
// 創(chuàng)建SparkStreaming接收kafka消息隊(duì)列數(shù)據(jù)的2種方式
// 一種是Direct approache,通過(guò)SparkStreaming自己主動(dòng)去Kafka消息隊(duì)
// 列中查詢還沒(méi)有接收進(jìn)來(lái)的數(shù)據(jù)，并把他們拉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))
// 采用迭代器的方式來(lái)避開(kāi)對(duì)象不能序列化的問(wèn)題竿刁。
// 通過(guò)對(duì)RDD中的每個(gè)元素實(shí)時(shí)產(chǎn)生推薦結(jié)果，將結(jié)果寫(xiě)入到redis麻献，或者其他高速緩存中们妥，來(lái)達(dá)到一定的實(shí)時(shí)性。
// 2個(gè)流的處理分成2個(gè)sparkstreaming的應(yīng)用來(lái)處理勉吻。
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 + "【實(shí)時(shí)】推薦了以下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()
}
}
步驟三：將創(chuàng)建的項(xiàng)目進(jìn)行打包上傳到服務(wù)器
mvn clean package -Dmaven.test.skip=true
步驟四：先編寫(xiě)個(gè)性化推薦代碼 shell 執(zhí)行腳本
[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
步驟五：執(zhí)行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>
步驟六：再編寫(xiě)model實(shí)時(shí)推薦代碼 shell 執(zhí)行腳本
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...
有任何問(wèn)題监婶，歡迎留言一起交流~~
更多文章：基于Spark的電影推薦系統(tǒng)：https://blog.csdn.net/liuge36/column/info/29285
