【电影推荐系统】实时推荐

概览

技术方案：

  • 日志采集服务：通过利用Flume-ng对业务平台中用户对于电影的一次评分行为进行采集，实时发送到Kafka集群。
  • 消息缓冲服务：项目采用Kafka作为流式数据的缓存组件，接受来自Flume的数据采集请求。并将数据推送到项目的实时推荐系统部分。
  • 实时推荐服务：项目采用Spark Streaming作为实时推荐系统，通过接收Kafka中缓存的数据，通过设计的推荐算法实现对实时推荐的数据处理，并将结构合并更新到MongoDB数据库。

1. 实现思路

我们应该如何实现？

  1. 首先应该redis安装，这里存储用户的第K次评分（用户评分存入redis中）
  2. 安装zookeeper，安装kafka，都是standlone模式
  3. 测试Kafka与Spark Streaming 联调。Kafka生产一条数据，Spark Streaming 可以消费成功,并根据redis中的数据和MongoDB数据进行推荐，存入MongoDB中
  4. 在业务系统写埋点信息，测试时写入本地文件，之后再远程测试写入云服务器log文件中
  5. flume配置文件书写，kafka创建两个topic，对整个过程进行测试

2 环境准备

1.1 redis 安装

  • redis安装redis安装
  • 密码：123456
  • 存入redis一些数据 lpush uid:1 mid:score
  • redis 教程：教程

1.2 zookeeper单机版安装

  • zookeeper安装：zookeeper安装
  • 版本：3.7.1
  • 遇到的坑：8080端口连接占用，我们需要在zoo.cpg文件中加上
    admin.serverPort=8001重新启动即可。

1.3 kafka单机安装

  • kafka安装：官网下载地址
  • 安装使用的为：127.0.0.1
  • 启动kafka：kafka教程
bin/kafka-server-start.sh config/server.properties
  • 创建一个topic
bin/kafka-topics.sh --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic recommender
  • 生产一个消息
bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic recommender
  • 消费一个消息
bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic recommender --from-beginning

3 测试kafka与spark streaming联调

  • kafka版本：2.2.0
  • spark版本：2.3.0
  • 因此使用spark-streaming-kafka-0-10

image.png

  1. 启动kafka，生产一条信息
  2. 书写程序
// 定义kafka连接参数val kafkaParam = Map("bootstrap.servers" -> "服务器IP:9092","key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" -> "recommender","auto.offset.reset" -> "latest")// 通过kafka创建一个DStreamval kafkaStream = KafkaUtils.createDirectStream[String, String]( ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String]( Array(config("kafka.topic")), kafkaParam ))// 把原始数据UID|MID|SCORE|TIMESTAMP 转换成评分流// 1|31|4.5|val ratingStream = kafkaStream.map{msg =>val attr = msg.value().split("\\|")( attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt )}
  1. 若是kafka报错，如果你同样也是云服务器，请注意kafka的配置信息（很重要！）

（1）解决方法：修改kafka配置文件，设置为设置listeners为内网ip，设置外网ip

  • 解决方案修改内网ip

（2）重新启动，成功

  • 内网外网分流：内网外网分流
  • kafka入门教程：入门教程
  1. redis报错：开启保护模式了，需要修改conf文件

效果

在kafka生产一个数据，可以在MongoDB中得到推荐的电影结果

4 后端埋点

前端进行评分后，触发click事件，后端进行测试埋点，利用log4j写入本地文件中。

4.1 本地测试

  • log4j配置文件
log4j.rootLogger=INFO, file, stdout# write to stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS}  %5p --- [%50t]  %-80c(line:%5L)  :  %m%n# write to file
log4j.appender.file=org.apache.log4j.RollingFileAppender
log4j.appender.FILE.Append=true
log4j.appender.FILE.Threshold=INFO
log4j.appender.file.File=F:/demoparent/business/src/main/log/agent.txt
log4j.appender.file.MaxFileSize=1024KB
log4j.appender.file.MaxBackupIndex=1
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS}  %5p --- [%50t]  %-80c(line:%6L)  :  %m%n
  • 埋点实现
//埋点日志
import org.apache.log4j.Logger;// 关键代码
Logger log = Logger.getLogger(MovieController.class.getName());
log.info(MOVIE_RATING_PREFIX + ":" + uid +"|"+ mid +"|"+ score +"|"+ System.currentTimeMillis()/1000)

4.2 写入远程测试

  1. Linux安装syslog服务，进行测试
  2. 主机log4j配置文件设置服务器ip
  • log4j配置：写入远程服务器
log4j.appender.syslog=org.apache.log4j.net.SyslogAppender
log4j.appender.syslog.SyslogHost= 服务器IP
log4j.appender.syslog.Threshold=INFO
log4j.appender.syslog.layout=org.apache.log4j.PatternLayout
log4j.appender.syslog.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS}  %5p --- [%20t]  %-130c:(line:%4L)  :   %m%n

5 flume配置

  1. flume对接kafka：flume对接文件
  2. flume设置source和sink，source为文件地址，sink为kafka的log
# log-kafka.properties
agent.sources = exectail
agent.channels = memoryChannel 
agent.sinks = kafkasink 
agent.sources.exectail.type = exec 
agent.sources.exectail.command = tail -f /project/logs/agent.log agent.sources.exectail.interceptors=i1 agent.sources.exectail.interceptors.i1.type=regex_filter agent.sources.exectail.interceptors.i1.regex=.+MOVIE_RATING_PREFIX.+ agent.sources.exectail.channels = memoryChannelagent.sinks.kafkasink.type = org.apache.flume.sink.kafka.KafkaSink agent.sinks.kafkasink.kafka.topic = log agent.sinks.kafkasink.kafka.bootstrap.servers = 服务器地址:9092 agent.sinks.kafkasink.kafka.producer.acks = 1 agent.sinks.kafkasink.kafka.flumeBatchSize = 20 agent.sinks.kafkasink.channel = memoryChannel
agent.channels.memoryChannel.type = memory
agent.channels.memoryChannel.capacity = 10000

6 实时推荐

ratingStream.foreachRDD{rdds => rdds.foreach{case (uid, mid, score, timestamp) => {println("rating data coming! >>>>>>>>>>>>>>>>")println(uid+",mid:"+mid)// 1. 从redis里获取当前用户最近的K次评分，保存成Array[(mid, score)]val userRecentlyRatings = getUserRecentlyRating( MAX_USER_RATINGS_NUM, uid, ConnHelper.jedis )println("用户最近的K次评分:"+userRecentlyRatings)// 2. 从相似度矩阵中取出当前电影最相似的N个电影，作为备选列表，Array[mid]val candidateMovies = getTopSimMovies( MAX_SIM_MOVIES_NUM, mid, uid, simMovieMatrixBroadCast.value )println("电影最相似的N个电影:"+candidateMovies)// 3. 对每个备选电影，计算推荐优先级，得到当前用户的实时推荐列表，Array[(mid, score)]val streamRecs = computeMovieScores( candidateMovies, userRecentlyRatings, simMovieMatrixBroadCast.value )println("当前用户的实时推荐列表:"+streamRecs)// 4. 把推荐数据保存到mongodbsaveDataToMongoDB( uid, streamRecs )}}
}

def computeMovieScores(candidateMovies: Array[Int],userRecentlyRatings: Array[(Int, Double)],simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Array[(Int, Double)] ={// 定义一个ArrayBuffer，用于保存每一个备选电影的基础得分val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()// 定义一个HashMap，保存每一个备选电影的增强减弱因子val increMap = scala.collection.mutable.HashMap[Int, Int]()val decreMap = scala.collection.mutable.HashMap[Int, Int]()for( candidateMovie <- candidateMovies; userRecentlyRating <- userRecentlyRatings){// 拿到备选电影和最近评分电影的相似度val simScore = getMoviesSimScore( candidateMovie, userRecentlyRating._1, simMovies )if(simScore > 0.7){// 计算备选电影的基础推荐得分scores += ( (candidateMovie, simScore * userRecentlyRating._2) )if( userRecentlyRating._2 > 3 ){increMap(candidateMovie) = increMap.getOrDefault(candidateMovie, 0) + 1} else{decreMap(candidateMovie) = decreMap.getOrDefault(candidateMovie, 0) + 1}}}// 根据备选电影的mid做groupby，根据公式去求最后的推荐评分scores.groupBy(_._1).map{// groupBy之后得到的数据 Map( mid -> ArrayBuffer[(mid, score)] )case (mid, scoreList) =>( mid, scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrDefault(mid, 1)) - log(decreMap.getOrDefault(mid, 1)) )}.toArray.sortWith(_._2>_._2)
}

7 启动顺序

  1. 启动hadoop、spark的容器
  • cd /docker
  • docker-compose up -d
  • docker-compose ps
  1. 启动mongodb和redis服务
  • netstat -lanp | grep "27017"
  • bin/redis-server etc/redis.conf
  1. 启动zookeeper、kafka服务
  • ./zkServer.sh start
  • bin/kafka-server-start.sh config/server.properties
  1. 启动flume服务
  • bin/flume-ng agent -c ./conf/ -f ./conf/log-kafka.properties -n agent

实现效果

前端评分成功后写入日志文件，flume对接log日志文件无问题，kafka对接flume无问题，spark streaming处理收到的一条数据，进行推荐，存入MongoDB中。

image.png

总结

由于时间匆忙，写的有些匆忙，如果有需要前端设计代码和后端的代码可以评论我，我整理整理发到github上。

前端设计部分没有时间去详细做，后续再对前端页面进行美化。本科当时整合了一个管理系统，现在也没有时间做，总之，一周多时间把当时的系统快速复现了下，算是一个复习。

在进行开发时，遇到许多问题，版本问题、服务器内网外网问题、docker容器相关问题、协同过滤算法设计问题，但帮着自己复习了下Vue和SpringBoot。

遇到问题时

  • 遇到问题不应该盲目解决，应该静下心看看报错原因，想想为何报错
  • 版本尤其重要，因此最好在一个project的pom设定版本
  • 使用服务器搭建docker-compose，利用该方法来搭建集群，快速简单，但涉及的端口转发等一些网络知识需要耐下心来看
  • Vue-Cli+Element-ui搭配起来开发简单
  • 写程序时，我们应该提前约定好接口，否则后续会很混乱…

后续

  • 后续将优化下前端页面，设计更多功能
  • 改进推荐算法
  • 增加冷启动方案

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/54783.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

echarts 图例组件legend配置

echarts 图例组件legend配置

legend 图例组件展示不同系列的图表类型标记、颜色、和名称。可以通过点击来控制哪个系列不展示。对于饼图来说&#xff0c;控制哪个数据不展示。 $> echarts5.4.0简单画一个饼图作为示例,设置legend:{show:true}展示图例。 const options {legend: {show: true,},series…
阅读更多...
【css问题】flex布局中，子标签宽度超出父标签宽度，导致布局出现问题

【css问题】flex布局中，子标签宽度超出父标签宽度，导致布局出现问题

场景&#xff1a;文章标题过长时&#xff0c;只显示一行&#xff0c;且多余的部分用省略号显示。 最终效果图&#xff1a; 实现时&#xff0c;flex布局&#xff0c;出现问题&#xff1a; 发现text-overflow: ellipsis不生效&#xff0c;省略符根本没有出现。 而且因为设置了 …
阅读更多...
《MySQL高级篇》十五、其他数据库日志

《MySQL高级篇》十五、其他数据库日志

文章目录 1. MySQL支持的日志1.1 日志类型1.2 日志的弊端 2. 慢查询日志(slow query log)3. 通用查询日志3.1 问题场景3.2 查看当前状态3.3 启动日志3.4 查看日志3.5 停止日志3.6 删除\刷新日志 4. 错误日志(error log)4.1 启动日志4.2 查看日志4.3 删除\刷新日志4.4 MySQL8.0新…
阅读更多...
Python实现GA遗传算法优化BP神经网络回归模型(BP神经网络回归算法)项目实战

Python实现GA遗传算法优化BP神经网络回归模型(BP神经网络回归算法)项目实战

说明&#xff1a;这是一个机器学习实战项目&#xff08;附带数据代码文档视频讲解&#xff09;&#xff0c;如需数据代码文档视频讲解可以直接到文章最后获取。 1.项目背景 遗传算法&#xff08;Genetic Algorithm&#xff0c;GA&#xff09;最早是由美国的 John holland于20世…
阅读更多...
zabbix监控mysql容器主从同步状态并告警钉钉/企业微信

zabbix监控mysql容器主从同步状态并告警钉钉/企业微信

前言&#xff1a;被监控的主机已经安装和配置mysql主从同步&#xff0c;和zabbix-agent插件。 mysql创建主从同步&#xff1a;http://t.csdn.cn/P4MYq centos安装zabbix-agent2&#xff1a;http://t.csdn.cn/fx74i mysql主从同步&#xff0c;主要监控这2个参数指标&#xf…
阅读更多...
用 docker 创建 jmeter 容器，能做性能测试？

用 docker 创建 jmeter 容器，能做性能测试？

我们都知道&#xff0c;jmeter 可以做接口测试&#xff0c;也可以用于性能测试&#xff0c;现在企业中性能测试也大多使用 jmeter。docker 是最近这些年流行起来的容器部署工具&#xff0c;可以创建一个容器&#xff0c;然后把项目放到容器中&#xff0c;就可以构建出一个独立的…
阅读更多...
【神经网络手写数字识别-最全源码（pytorch）】

【神经网络手写数字识别-最全源码（pytorch）】

Torch安装的方法 学习方法 1.边用边学&#xff0c;torch只是一个工具&#xff0c;真正用&#xff0c;查的过程才是学习的过程2.直接就上案例就行&#xff0c;先来跑&#xff0c;遇到什么来解决什么 Mnist分类任务&#xff1a; 网络基本构建与训练方法&#xff0c;常用函数解析…
阅读更多...
Qt 添加资源文件（添加图片）

Qt 添加资源文件（添加图片）

第一步&#xff0c;先将要添加的资源文件&#xff08;图片等&#xff09;放入项目目录&#xff0c;如下&#xff1a; 第二步&#xff0c;如下图&#xff1a; 第三步&#xff0c;如下图&#xff1a; 第四步&#xff0c;如下图&#xff1a; 第五步&#xff0c;继续选择完成即…
阅读更多...
ES6 - 对象新增的一些常用方法

ES6 - 对象新增的一些常用方法

文章目录 1&#xff0c;Object.is()2&#xff0c;Object.asign()3&#xff0c;Object.getOwnPropertyDescriptors()4&#xff0c;Object.setPrototypeOf()和getPrototypeOf()5&#xff0c;Object.keys()、values() 和 entries()6&#xff0c;Object.fromEntries()7&#xff0c;…
阅读更多...
学习系统编程终章【多线程剩余知识】

学习系统编程终章【多线程剩余知识】

引言&#xff1a; 北京时间&#xff1a;2023/8/3/19:21&#xff0c;刚刚将文章更新&#xff0c;是近期以来为数不多的一次早更&#xff0c;不然每次更文都要卡在12点左右&#xff0c;这是我们实现日更的一个好开端&#xff0c;再接再厉实现日更不是梦。最近失眠一直困扰着我&a…
阅读更多...
参考RabbitMQ实现一个消息队列

参考RabbitMQ实现一个消息队列

文章目录 前言小小消息管家1.项目介绍2. 需求分析2.1 API2.2 消息应答2.3 网络通信协议设计 3. 开发环境4. 项目结构介绍4.1 配置信息 5. 项目演示 前言 消息队列的本质就是阻塞队列&#xff0c;它的最大用途就是用来实现生产者消费者模型&#xff0c;从而实现解耦合以及削峰填…
阅读更多...
Java项目作业~ 创建基于Maven的Java项目，连接数据库，实现对站点信息的管理，即实现对站点的新增，修改，删除，查询操作

Java项目作业~ 创建基于Maven的Java项目，连接数据库，实现对站点信息的管理，即实现对站点的新增，修改，删除，查询操作

需求&#xff1a; 创建基于Maven的Java项目&#xff0c;连接数据库&#xff0c;实现对站点信息的管理&#xff0c;即实现对站点的新增&#xff0c;修改&#xff0c;删除&#xff0c;查询操作。 以下是站点表的建表语句&#xff1a; CREATE TABLE websites (id int(11) NOT N…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023