【电影推荐系统】数据爬取、数据加载进MongoDB数据库

概览

本篇主要介绍数据来源、数据加载进数据库过程

1 数据获取

使用Scrapy爬取豆瓣电影数据，然后利用movielens数据集来造一份rating数据。

1.1 数据集获取

数据集获取：选取movielens 数据集：movielens官网
数据集包括：movies,ratings,tags文件

1.2 数据爬取

使用scrapy+xpath爬取豆瓣电影数据，最后存入csv，命名为movie.csv
对于爬取的数据进行预处理：包括字段选取、相关字符处理

1.3 数据转换

由于缺少rating数据，因此我们使用movielens的rating文件来造一下评分数据。
movielens数据集movie文件一共2791条电影数据，因此我们直接截取爬取的movie.csv前2791条数据。
直接用movielens的电影ID替换爬取movie.csv的电影ID，因此最后我们得到的电影数据也就有了对应的评分数据。
最后，我们需要的只是两个文件：movie.csv,rating.csv

（1）Movie数据

数据表格式为：
mid,title,desc,minute,year,year,language,geners,actors,director

（2）Rating数据

userID,mid,score,timestamp

2 将数据加载进MongoDB数据库

我们选择MongoDB数据库的原因如下：

千万级别的文档对象，近10G的数据，对有索引的ID的查询不会比mysql慢，而对非索引字段的查询，则是全面胜出
可以进行深度查询

接下来，我们在云服务器部署MongoDB，主机远程连接数据库，将文件加载进数据库中。

2.1 MongoDB安装

安装教程：linux安装MongoDB

2.2 maven依赖

maven相关依赖版本如下

注意：Spark版本和Spark集群的版本需要一致

scala:2.11.8
Spark:2.3.0

<properties><scala.version>2.11.8</scala.version>
</properties><dependencies><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>2.3.0</version></dependency>
</dependencies>

2.3 数据加载的程序

// 加载数据主程序
object DataLoader {val MONGODB_MOVIE_COLLECTION = "Movie"val MONGODB_RATING_COLLECTION = "Rating"val mgo_host = "root"val config = Map("spark.cores" -> "local[*]","mongo.uri" -> "mongodb://root:123456@服务器公网IP:27017/recommender","mongo.db" -> "recommender")// 文件位置val MOVIE_DATA_PATH = "F:\\1-project\\offline\\src\\main\\resources\\file\\movie.csv"val RATING_DATA_PATH = "F:\\1-project\\offline\\src\\main\\resources\\file\\ratings.csv"def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("DataLoader")val spark = SparkSession.builder().config(sparkConf).getOrCreate()import spark.implicits._val movieRDD = spark.sparkContext.textFile(MOVIE_DATA_PATH)  // 加载数据// 转为dfval movieDF = movieRDD.map(item => {val attr = item.split(",")Movie(attr(0).toInt, attr(1).trim, attr(2).trim, attr(3).trim, attr(4).trim, attr(5).trim, attr(6).trim, attr(7).trim, attr(8).trim)}).toDF()val ratingRDD = spark.sparkContext.textFile(RATING_DATA_PATH)val ratingDF = ratingRDD.map(item => {val attr = item.split(",")Rating(attr(0).toInt,attr(1).toInt,attr(2).toDouble,attr(3).toInt)}).toDF()implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))// 将数据保存到MongoDBstoreDataInMongoDB(movieDF, ratingDF)spark.stop()}}

def storeDataInMongoDB(movieDF: DataFrame, ratingDF: DataFrame)(implicit mongoConfig: MongoConfig): Unit ={// 新建一个mongodb的连接val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))// 将DF数据写入对应的mongodb表中movieDF.write.option("uri", mongoConfig.uri).option("collection", MONGODB_MOVIE_COLLECTION).mode("overwrite").format("com.mongodb.spark.sql").save()ratingDF.write.option("uri", mongoConfig.uri).option("collection", MONGODB_RATING_COLLECTION).mode("overwrite").format("com.mongodb.spark.sql").save()//对数据表建索引mongoClient(mongoConfig.db)(MONGODB_MOVIE_COLLECTION).createIndex(MongoDBObject("mid" -> 1))mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION).createIndex(MongoDBObject("uid" -> 1))mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION).createIndex(MongoDBObject("mid" -> 1))mongoClient.close()}