基于Hadoop与Spark大数据平台的个性化图书推荐系统搭建学习总结_基于spark的图书推荐系统实验报告

前言：这两个月来一直在处理接手实验室师兄的一个图书推荐项目，期间从读懂其用python构建的简易推荐系统到在spark上写pyspark、scala程序来实现一个基于大数据平台的分布式推荐系统，对于我这样一个无人指点的小白着实是费了一番功夫，现在做记录如下。

一、在spark分布式平台运到的坑

1、 如何在spark ui上监听到spark的历史运行记录

利用spark UI 调试和监控运行的spark程序非常的方便，它可以让我们很直观的看到正在运行的stages、某stage运行的时间和executors的运行状况等，但有个比较不便的是没用经过特殊设置，spark UI 在spark程序运行完就不能再连接上了。市场上常见的spark书籍在教搭建spark平台时都不会教怎么配置spark环境变量以使我们可以看到spark平台运行的历史记录。在网上进行搜索一番后发现如何配置环境也都是众说纷纭，不同的人配置的方式也不用。
经过一番比较后，我配置spark的环境变量以使其可以看到spark的历史运行记录的方式如下：
在环境变量文件spark-defaults.conf中添加如下几行：

spark.history.ui.port=Master:18080   //设置监听历史记录的端口号为18080
spark.eventLog.enabled=true   //设置为ture，表示设置记录Spark事件，用于应用程序在完成后重构webUI
spark.eventLog.dir=hdfs://Master:9000/tmp/spark/events   //设置hdfs缓存历史的任务目录，目录应在hdfs中相应的创建好
spark.history.retainedApplications=30  //设置可查看的最大历史记录数
1
2
3
4

在环境变量文件spark-env.sh中添加如下几行：

export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=18080 -Dspark.history.retainedApplications=3 -Dspark.history.fs.logDirectory=hdfs://Master:9000/tmp/spark/events" //设置后相应hdfs目录需相应的创建好
1

参考连接官方文档，博客1
在上述两个环境变量文件中添加好相应代码后，在启动spark集群时同时打开spark-history-server.sh脚本即可在spark UI中监听到spark 的历史运行记录了。

2、 用spark-submit提交pyspark代码没有按预期运行分布式模式

异常描述： 分布式平台（hadoop+spark）配置好后，我在spark-shell master spark://Mastter:7070入口进入交互式环境下和以spark-submit为入口在yarn集群和spark standalone 集群模式上提交sbt打包的scala程序都是以运行分布式模式运行程序的。但在spark-submit为入口，按书上的提示在yarn集群和spark standalone 集群模式上提交pyspark.py程序欲使其运行分布式时就出现了问题，在运行记录中看到的是提交的pyspark代码是以localhost模式形式运行的，也就是说跑的是单机。这个问题困扰我两天，因为我的spark和hadoop是照着几本书配置的，也参考了师兄配置的集群且提交打包好的jar程序运行时并无异常。期间在网上查阅许久，也把spark和hadoop的配置文件检查了数遍，及在一些技术群里询问但都没有得到答案，最后问题的解决是来自于一个无意的尝试。平常在spark-submit提交的pyspark程序的格式是 “spark-submit /opt/data/LibraryProject/programSet/base_user_submit.py(程序名) --master yarn --deploy-mode client --queue default --executor-memory 5g --num-executors 3 “后来改为“spark-submit --master yarn --deploy-mode client --queue default --executor-memory 4g --num-executors 3 --executor-cores 4 --driver-memory 1G /opt/data/LibraryProject/programSet/base_user_submit.py”即将代码的的位置放在提交指令的最后一栏就可以如愿在yarn上运行分布式程序了。
　　异常分析：经过分析后发现，spark-submit提交方式应该是读到所提交的程序文件就不会再读后面如–master yarn --deploy-mode client…等提交的参数设置了，而我所提交的程序中在创建spark实例的时候刚好没有进行任何给参数的操作（我创建spark的方式是spark=SparkSession(context),context=SparkContext()），所以出现这样spark平台即以默认提交方式localhost形式运行程序了。这真的是一个大坑，现阶段介绍pyspark的书比较少，且介绍的较浅，尽管python在越来越多的应用到编写spark程序实现分布式任务当中。

3、spark在执行程序中出现too much file open 的错误

写在我的另一个博客中
Uncaught exception while reverting partial writes to file …(Too many open files)

4、运行pyspark程序运到python版本问题

问题分析：因为在集群中存在python官网中下载安装的python和安装的anaconda带的python两个版本，程序在执行时不知道使用哪个版本python的解释器。
　　
　　问题解决：
在环境变量文件spark-env.sh中添加如下设置：

export PYSPARK_PYTHON=/opt/app/anaconda2/bin/python

1
2

二、 基于读者与读者的图书推荐实现

1、基于用户的协同过滤的基本思想是：根据用户对物品的爱好找到相邻邻居用户，然后将邻居用户偏爱的物品中但是当前用户又没有购买的推荐给他。它的原理就是将某个用户对全部物品的爱好作为一个向量，计算出各个用户之间的相似度，找到邻居后，依据邻居们的相似度及购物历史，预测当前用户可能会喜欢的但尚未购买的物品，计算得出按照一定排列顺序的物品推荐类表给用户。
2、基于用户的算法在很多方面有着非常成功的运用，但是随着推荐系统规模的日益扩充，特别是用户数和项目数的日益增多，该算法存在的弱点开始体现出来了。
a、稀疏性问题。据研究结果表明，当用户评价项目数少于总项目数的，就很容易造成评价矩阵数据相当稀疏，导致算法难以找到一个用户的偏好相似邻居。
b、冷启动问题。基于用户协同过滤是建立在有大量用户对某个产品的评价上的，由于在新产品开始阶段没有人购买，也没有对其进行评价，那么在开始阶段也将无法对其进行推荐。
c、算法扩展性问题。随着物品数尤其是用户数的剧烈增加，最近邻居算法的计算量也相应增加，所以不太适合数据量大的情况使用，所以推荐系统性能也会大大受影响，而现在的推荐系统几乎是结构，没有快速的相应速度，对网络用户来说无法忍受的，因此这在某种程度上限制了基于用户协同过滤在推荐系统中的使用。
　d、特殊用户问题。在生活中，往往有一部分人的偏好是比较特殊，他没有相对固定的兴趣爱好，而这刚好是基于用户协同过滤的前提，那么系统很难为他找出邻居，也就是很难给出比较精确的推荐信息了。
3、基于读者与读者的协同过滤如何实现
a、构建读者-图书评价矩阵
b、利用余弦相似度计算公式，以每个读者的借阅评分向量来计算不同用户之间的相似度。
c、制定推荐策略，为每一位读者推荐图书

4、在实现过程中遇到的理论问题
　　图书馆没有对图书进行借阅打分的机制，因而在构建用户-图书评价矩阵时缺失读者借阅评分数据，只能以简单的是否借阅某书来代替读者对某书的评分。其中借阅过表示为1，代表完全满意；没有借阅过表示为0，代表完全不满意。这种无可奈何的办法会导致不能准确计算读者的相似度，且容易造成推荐热门借阅图书，难以对读者进行个性化推荐。所以如果想要实现对读者进行个性化推荐，后期图书馆应该制定一个可以让读者对其所借阅的图书进行评分的机制。

三、 代码中觉得值得记录的操作

1、读取csv文件并将其转化为dataframe的操作指令

user_book=spark.read.csv('logSet/user_book.csv',inferSchema='true',header="true")//pyspark版，读取时设置为自动推断列的类型，取第一列为列名
var user_book_list=spark.read.format("csv").option("header","true").load("dataSet/user_book_id.csv")//scala版本，读取时设置为自动推断列的类型，取第一列为列名
1
2

2、保存datafram到hdfs（df.repartition(1)表示将数据聚合在一个分区上进行保存，这样保存的数据不会别切分成多个）

book_list.write.repartition(1).csv('test_book.csv',header='true',mode='overwrite')//pyspark版，保存时时设置未取第一列为列名，如有相同文件名则进行覆盖
user_sim.repartition(1).write.format("csv").option("header","true").option("mode","overwrite").save("spark_result/user_Sim3.csv")scala版，保存时时设置未取第一列为列名，如有相同文件名则进行覆盖
1
2

3、创建dataframe的几种方式
pyspark版

schema = StructType([StructField("user_id", IntegerType(), True), StructField("sim_id", IntegerType(), True),StructField("sim", FloatType(), True)])
df=spark.createDataFrame([(user_id,i,sim_values)],schema)
1
2

scala版
a、创建带制定格式的dataframe

case class userSim(user_id:String,sim_id:String,sim_value:Double)
var df_tmp = List(userSim(user_id,user_i,sim_value))
var user_sim=sc.parallelize(df_tmp).toDF()
或者：
var list_tmp = List(userSim(user_book_cv_list(0)(0).toString,  user_book_cv_list(0)(0).toString, 1.0)).toBuffer//预设df的格式
list_tmp += userSim(user_book_cv_list(i)(0).toString,  user_book_cv_list(j)(0).toString, cosineSim)
val user_sim = sc.parallelize(list_tmp).toDF()//将得出的读者相似度list创建为df(user_id,sim_id,sim_value)
1
2
3
4
5
6
7

b、不带格式的dataframe

val df = Seq(("00",Array("机器学习","深度学习","spark机器学习"))).toDF("user_id","userName")
var arr =("大数据导论","快速大数据分析")
val df4 = Seq(("02",Array(arr))).toDF("user_id","userName")
1
2
3

4、合并两个dataframe(在代码实现过程的用到多次，很好用)

user_book=user_book.join(book_id,"bookName","left").cache()//pyspark版，以两个df都有的col（bookName）进行合并，合并模式是"left"，数据向df user_book倾斜
var user_book_list=user_book.join(book_list,$"bookName"===$"bookName1","left")//scala版
var user_sim_list = sim_list.join(all_user,sim_list("sim_id")===all_user("user_id"),"inner")//scala版，合并两个df中col(sim_id)===col(user_id)相同的值，，join的模式是"inner"，注意用的是三个“=”
1
2
3

5、将某列值相同的标签聚合到同一行

var all_user = all_user.withColumn("bookName1", $"bookName").groupBy("user_id").agg(collect_list($"bookName1")).toDF("user_id","bookName").persist(StorageLevel.MEMORY_ONLY)/以读者ID聚合，得到一个型为(user_id,Array(user_all_book))的dataframe
1

6、删除col(1)中包含col(0)中具有的值（这个在实现对每个读者进行最终推荐图书时非常好用，而且实现起来很简单）

//自定义一个函数
val filterList = udf((a: Seq[String], b: Seq[String]) => a diff b)
//在col(recomBook)去除掉user_id用户借阅过的书籍
var user_sim_recomBook=user_sim_book.withColumn("left", filterList($"recomBook", $"user_bookName")).show()
1
2
3
4

7、构建读者-评分矩阵，（这里用了一个取巧的方法–调用分词函数，处理起来也很快，120万行的数据只需要几分钟。实验室用同学写过类似的代码，借鉴了一部分。不过这里可以这么用刚好是因为除处理的图书没有评分数据，只能以借阅与否代表满意度，否则不能直接用下示方法）

import org.apache.spark.ml.feature.CountVectorizer
import org.apache.spark.storage.StorageLevel

var user_book_list=spark.read.format("csv").option("header","true").load("dataSet/user_book_id.csv").select("id","bookName","book_id").toDF("user_id","bookName","book_id")//加载全部读者的图书借阅记录
user_book_list.persist(StorageLevel.MEMORY_ONLY)//相当于调用cache()函数
var all_user=user_book_list.select("user_id","bookName").cache()//取用户名和用户借阅书名
val user_book = all_user.withColumn("bookName1", $"bookName").groupBy("user_id").agg(collect_list($"bookName1")).toDF("user_id","bookName").cache()//以读者ID聚合，得到一个型为(user_id,Array(user_all_book))的dataframe
val book_cv = new CountVectorizer().setInputCol("bookName").setOutputCol("TitleVectors")//以待分词的列为输入创建分词实例
val bookmodel = book_cv.fit(user_book)//调用fit实例
val user_book_cv = bookmodel.transform(user_book)//得到形为(user_id,bookName,TitleVectors)的dataframe，其中的col(TitleVectors)为每个读者的读者-评分矩阵
1
2
3
4
5
6
7
8
9
10

四、低效率编写代码的体现（spark平台用的好的人可以实现快速计算数以Tb的数据，但代码没有优化好，写出来的程序可能还不如用pyython单机来的快,这一点自己深有体会，以下是自己遇到的一些坑）

1、不会学会看spark官方文档。
2、重复使用active操作，特别是在大的循环里。
3、不会为重复调用的rdd或者df设置缓存。
4、想实现某一功能，特别是spark官方文档没有直接写到的，不是先googel而是自己硬着写。
5、写代码之前没有提前都构思好，想到什么写什么，写到后面才发现方向错了。
6、能用scala实现的功能要用pyspark实现。官方指出，pyspark代码执行的速度比scala慢一倍(rdd)。

五、项目处理过程中查阅的觉得不错的博客

1、How to create an empty DataFrame? Why “ValueError: RDD is empty”?
2、How to create an empty DataFrame with a specified schema?
3、Spark DataFrame按某列降序排序](https://blog.csdn.net/dkl12/article/details/80961981)
4、Python pyspark.sql.SparkSession() Examples
5、Merging multiple data frames row-wise in PySpark
6、Extract column values of Dataframe as List in Apache Spark
7、spark - Converting dataframe to list improving performance
8、How do I add a new column to a Spark DataFrame (using PySpark)?
9、PySpark 学习笔记三
10、spark dataframe 一列分隔多列，一列分隔多行（scala）
11、PySpark︱DataFrame操作指南：增/删/改/查/合并/统计与数据处理
12、Trying to create dataframe with two columns [Seq(), String] - Spark
13、how to convert rows into columns in spark dataframe, scala
14、Difference between columns of ArrayType in dataframe
15、python—pandas.merge使用