我的基于Spark的电影推荐（呆板学习实训）

基于Spark的电影推荐

一、实训背景

近几年互联网信息量呈几何级增长，用户很轻易迷失在海量信息中。虽然可使用层次分类（分类目录）或搜索引擎等方法解决这类信息过载问题，但是层次分类需要手工对信息举行分类，并且随着数据量增长层级渐渐增多，不便于用户浏览；而搜索引擎需要用户自己输入关键词，自行选择结果，这要求用户已明白知道要查找的内容，并且具有核心词的抽象本事，如果对结果不满意需要调整关键词重新搜索。推荐体系根据用户的浏览记载、交际网络等信息举行个性化的计算，发现用户的兴趣，并应用推荐算法最终到达“千人千面”“个性化”推荐的结果。
二、实训内容

使用Spark的呆板学习算法实现库MLlib，以及将其应用在MovieLens数据集上实现个性化电影推荐。
三、关键技能

• 基于协同过滤的推荐算法原理
  
• 最小二乘法（ALS）原理
  
• pyspark推荐算法模型练习
  

四、实训目的

• 了解推荐体系的应用场景
  
• 熟悉协同过滤推荐算法原理
  
• 把握pyspark中使用最小二乘法（ALS）来实现协同过滤推荐模型。
  

五、实训环境

• 操纵体系：ubuntu16
  
• 工具软件：jupyter notebook、Python 3.6.13
  
• 硬件环境：无特别要求
  
• 核心库：
  
  
  
  • pyspark 3.0.1
    
  
  

六、实训原理

1 推荐体系的应用场景

• 电商平台
  现在推荐体系已经根本成为电商平台标配。主流的电商平台具有多种推荐形式，比方“猜你喜欢”“购买此商品的用户也购买了……”等；除 此之外，还有隐式商品推荐，比方在搜索结果中将推荐商品排名提前。
  
• 个性化视频网站
  每年国内外都有大量电影上映，由用户自制的视频节目也越来越多，用户很难在海量的视频节目中举行选择。视频网站基于用户的汗青观看记载以及视频内容之间的内涵联系，分析用户潜伏兴趣，向用户推荐其感兴趣的内容。
  
• 音乐歌单
  现在音频类个性化推荐主要是向用户推荐歌曲或播单，好的推荐会让用户既熟悉又有新鲜和惊喜的感觉。主流音乐平台音乐推荐的实现方法与电影推荐类似，主要基于音乐的风格、用户收听汗青、用户收听行为等举行协同过滤。
  
• 交际网络
  推荐体系在交际网络中的应用主要是挚友推荐和内容推荐。挚友推荐是指在交际网站中向用户推荐具有共同兴趣的用户成为挚友。用户之间可通过关系网络创建联系，还可以通过阅读、点赞、批评了雷同的博文产生关系。如果两个用户有多个共同的标签，曾经批评大概转发雷同的信息，阐明他们对这条信息有着共同的兴趣。对这些用户行为应用基于用户的协同过滤算法，就可以向用户举行个性化内容推荐。在交际网站中用户之间形成一个交际网络图，可以分析用户之间兴趣的相似性，比方，用于学术社区中偕行的发现，对那些研究领域雷同，但在网站中并非挚友的用户，推荐他们互加挚友。
  
• 新闻网站
  新闻网站中应用推荐算法可以方便用户实时获取个性化信息，减少用户浏览、检索新闻的时间，并提供更好的阅读体验，从而增长用户黏性。一样平常采用基于内容的协同过滤推荐算法来实现，其中数据包括用户属性特性、浏览汗青和新闻内容等，从而解决新闻量过大时给用户带来的信息过载和迷航问题。在新闻网站中常有“冷启动”的问题。它是指网站刚刚创建，用户和 新闻内容较少，用户的行为数据更少，所以协同过滤算法往往无效。为缓解此问题，可以使用热门内容作为推荐结果，渐渐收集用户行为数据，不断美满推荐结果，吸引更多用户注册，从而形成良性循环。
  
• 个性化阅读
  个性化阅读是为每一位用户定制其感兴趣的个性化内容，比方新闻、论坛帖子、小说等。推荐体系通过推荐算法得到用户兴趣，并向其推送个性化的阅读内容，从而提供更优的阅读方式和更好的阅读体验。
  
• 个性化广告
  个性化广告是指有针对性地向特定用户展示特定广告内容。起首对广告受众举行用户画像，对受众的个人状态、贸易兴趣、交际图谱等方面举行刻画，这是广告推荐引擎的根本。然后推荐体系基于用户的行为举行协同过滤，并对推荐的广告结果举行粗选、精选。上述过程可以通过定时运行的方式离线生成推荐结果；也可以实时生成推荐结果，这对推荐算法和硬件计算均要求较高。在用户浏览广告过程中可看出用户对广告的态度或反应，这可作为推荐结果的评价依据，并用于改进推荐算法，减少用户对广告的负面体验。
  

2 Spark介绍

Spark是一个开源的并行计算与分布式计算框架，最大特点是基于 内存计算，适合迭代计算，兼容Hadoop生态体系中的组件，同时包括相 关的测试和数据生成器。其计划目的是全栈式解决批处置惩罚、布局化数据查询、流计算、图计算和呆板学习等应用，适用于需要多次操纵特定数据集的应用场所。需要反复操纵的次数越多，所需读取的数据量越大，服从提拔越大，在这方面比Hadoop快很多倍。Spark集成了以下模块， 为差别应用领域的从业者提供了快速的大数据处置惩罚方式。

• Spark SQL：分布式SQL查询引擎，提供了一个DataFrame编程抽 象。
  
• Spark Streaming：把流式计算分解成一系列短小的批处置惩罚计算，并 提供高可靠和吞吐量服务。
  
• MLlib：Spark对常用的呆板学习算法的实现库，支持4种常见的机 器学习问题：分类、回归、聚类和协同过滤。
  
• GraphX：提供图计算服务。
  
• SparkR：支持R语言的库。
  
• 扩展库pyspark：提供了SparkContext作为主要入口，还有弹性分布式数据集（Resilient Distributed Dataset，RDD）、文件访问类（SparkFiles）等可以公开访问的类，并且提供了pyspark.sql、pyspark.streaming与pyspark.mllib等模块与包。
  

3 基于协同过滤的推荐算法介绍

基于用户生齿属性和行为数据计划的推荐算法，称为协同过滤算法。此方法主要根据用户的汗青行为，寻找用户或物品的近邻集合，以此计算用户对物品的偏好，包括基于领域、图、关联规则、知识的推荐算法，其中最广泛应用的是基于领域的方法，在实践中往往是上述几种方法的混合应用。
基于领域的推荐算法分类：
基于领域的推荐算法主要包含两种：基于用户的协同过滤算法
（UserCF）和基于物品的协同过滤算法（ItemCF）。基于用户的协同过滤计算用户兴趣相似度，基于物品的协同过滤算法计算与用户偏好的物品相似的物品。本实行电影推荐使用的是基于用户的协同过滤算法。
基于用户的协同过滤算法：
基于用户的协同过滤算法为用户推荐兴趣相似的其他用户喜欢的物品。算法的关键是计算两个用户的兴趣相似度。计算用户相似度的常用方法有余弦相似性、皮尔森系数相关和修正的余弦相似性。
算法步调如下：
① 找到与目的用户兴趣相似的用户集合；
② 找到这个集合中的用户喜欢的，且目的用户没有用过的物品， 推荐给目的用户。
4 基于用户的协同过滤推荐示例

用户/物品物品A物品B物品C物品D用户A√√推荐用户B√用户C√√√ 该表是基于用户的协同过滤推荐示例，可以看到用户A与用户C 所喜欢的物品具有较多的交集，即两个用户具有相似性，那么用户C喜 欢的物品很有可能用户A也会喜欢，而用户C喜欢物品D，则可以向用户 A推荐物品D。
计算用户兴趣相似度时，要制止热门物品自带马太效应的影响，即大部门用户可能都对热门的物品表现出喜欢的状态，但是这些用户之间并非一类人，因为所谓的热门物品区分度较弱。
基于用户的协同过滤算法的缺点是随着用户数目增大，计算用户兴趣相似度越来越复杂，时间和空间复杂度与用户数接近于平方关系。所以一样平常采用离线方式举行推荐，即当用户产生新的行为时，不会立即举行计算，所以推荐结果并不会立刻发生厘革。别的，这一算法是基于隐式群体的兴趣举行推荐，可解释性不强。这一算法适用于用户兴趣比较稳固的场景，即通过群体的兴趣来代表用户个体的兴趣，一旦群体的兴趣确立，就可以认为个体用户服从此兴趣，由此向其举行推荐，结果一样平常较准确。
5 模型练习(ALS算法)

在pyspark中使用交替最小二乘法（Alternating Least Squares， ALS）来实现协同过滤推荐，主要缘故起因是它支持稀疏的输入数据（用户 对物品的评分是稀疏矩阵），并且可用简朴的线性代数运算求解最优解，别的，输入数据本身可以并行化，这就使ALS在大规模数据上速度 非常快。
ALS中文名作交替最小二乘法，在呆板学习中，ALS特教唆用最小二乘法求解的一个协同过滤算法，是协同过滤中的一种。ALS算法是2008年以来，用的比较多的协同过滤算法。它已经集成到Spark的Mllib库中，使用起来比较方便。从协同过滤的分类来说，ALS算法属于User-Item CF，也叫做混合CF，因为它同时考虑了User和Item两个方面，即可基于用户举行推荐又可基于物品举行推荐。
一样平常而言用户只会购买物品会集的少少数部门产品，并对其举行打分。考虑下面如许一个包含用户的打分矩阵（列为用户u1-u6，行为物品I1-I8），我们可以看到这个用户的评分矩阵是十分稀疏的，有很多用户的购买的记载是空的，而且在现实业务中，用户的评分矩阵会更加的稀疏。怎样通过如许一个稀疏矩阵，对用户举行协同推荐用户可能很喜欢的物品对于推荐体系而言是一种很大的磨练。
用户评分矩阵
l1l2l3l4l5l6l7l8u152u2431u315u472u5711u6542 解决稀疏矩阵问题，需要采用矩阵分解，如下面两个图：
将原本矩阵A(m * n) 分解成X(m * rank) 矩阵与Y(rank * n) 矩阵，而且A大约即是 X * Y

在spark MLlib 呆板学习库中现在推荐模型只包含基于矩阵分解(matrix factorization)的实现。具体的分解思绪，找出两个低维的矩阵，使得它们的乘积是原始矩阵。因此这也是一种降维技能。假设我们的用户和物品分别是m和n，那对应的“用户-物品”矩阵A，类似图所示：

找到和“用户-物品“矩阵近似的k维（低阶）矩阵，最终照旧要求出如下两个矩阵：一个用于表示用户X维矩阵，以及一个表征物品的Y维矩阵。这两个矩阵也称为因子矩阵，他们的矩阵乘积便是原始评级数据的一个近似值。值得注意的是，原始评级矩阵通常很稀疏，但因子矩阵却是稠密的，如图所示：

ALS是求解矩阵分解问题的一种最优化方法，它功能强大，结果理想而且被证明相对轻易实现。这使得它很适合如Spark如许的平台。
ALS的实现原理是迭代式求解一系列最小二乘回归问题。在每次迭代时，固定用户因子矩阵大概是物品因子矩阵中的一个，然后用固定的这个矩阵以及评级数据来更新另一个矩阵。之后，被更新的矩阵被固定住，再更新另外一个矩阵。云云迭代，知道模型收敛（大概是迭代了预设好的次数）。
6 电影数据集

实行采用MovieLens数据集作为数据源，它是一个关于电影评分的 数据集。包括links.csv、movies.csv、ratings.csv、tags.csv几个文件。
link.csv文件的内容是电影编号，通过编号可以在网站上找到对应的电影链接；

• movies.csv文件中包含电影编号、标题、电影题材；
  
• ratings.csv文件是用户对电影的评分和评分时间戳，其中评分是5分制，按半星的 规模递增；
  
• tags.csv文件包含用户对电影的标签化评价和打标签时间戳；上述数据的编号、时间戳等均为数字型（long型），评分为float型。
  
• MovieLens数据集按照数据量的巨细分为10万、2000万yue 、2600万等 几种压缩包，方便差别用途的应用，其中10万的数据量较少，包括10万 条评分记载、1300个标签，对应9000部电影，用户数是700人，更新时间是2016年10月；2000万的压缩包有2000万条评分记载，565000个标 签，对应2.7万部电影，用户数是13.8万人，并且包括标签genome数据； 2600万的压缩包是全量数据，包含2600万条评分数据、75万个标签、 4.5万部电影、27万用户数，并且还有1200万条标签genome数据，末了 更新时间是2017年8月份。本次实行过程中采用10万的数据会集的rating.csv举行验ratings数据。
  

文件里面的内容包含了每一个用户对于每一部电影的评分。数据格式如下：userId, movieId, rating, timestamp；

• userId: 数字范例的每个用户的编号
  
• wget http://files.grouplens.org/datasets/movielens/ml-100k.zip: 数字范例每部电影的编号
  
• rating: 浮点范例的用户评分值，是5星制，按半颗星的规模递增(0.5 stars - 5 stars)
  
• timestamp: 评分时间戳，本实行中不使用。
  

数据排序的顺序按照userId，movieId排列的。
我们使用pandas观察数据集的形貌及前五行：

1. import os
2. os.environ["PYSPARK_PYTHON"]="/usr/bin/python3"
3. os.environ["PYSPARK_DRIVER_PYTHON"]="/usr/bin/python3"
4. # 数据简介
5. import pandas as pd
6. ratings = pd.read_csv('../dataset/ratings.csv',header=None,names=['userid','movieid','rating','timestamp'])
7. ratings.describe()

复制代码

  useridmovieidratingtimestampcount100004.000000100004.000000100004.0000001.000040e+05mean347.01131012548.6643633.5436081.129639e+09std195.16383826369.1989691.0580641.916858e+08min1.0000001.0000000.5000007.896520e+0825%182.0000001028.0000003.0000009.658478e+0850%367.0000002406.5000004.0000001.110422e+0975%520.0000005418.0000004.0000001.296192e+09max671.000000163949.0000005.0000001.476641e+09

1. # 前5行
2. ratings.head(5)

复制代码

  useridmovieidratingtimestamp01312.512607591441110293.012607591792110613.012607591823111292.012607591854111724.01260759205 七、实训步调

1 导入库

1. import os
2. import math
3. import time
4. from pyspark import SparkContext
5. from pyspark.sql import SQLContext, Row, SparkSession
6. from pyspark.mllib.recommendation import ALS

复制代码

2 加载数据库

起首对数据举行预处置惩罚，加载评分文件，并将文件中的记载按照6:2:2分为练习集、验证集、测试集，随机数种子固定为10。

1. sc = SparkContext()
3. #文件访问
4. small_raw_data = sc.textFile('../dataset/ratings.csv')
5. small_data = small_raw_data.map(lambda line: line.split(",")).map(lambda col: (col[0], col[1], col[2])).cache()
8. #按照6:2:2分为训练集、验证集、测试集
9. training_RDD, validation_RDD, test_RDD = small_data.randomSplit([6, 2, 2], seed=10)
10. validation_predict_RDD = validation_RDD.map(lambda x: (x[0], x[1]))
11. test_predict_RDD = test_RDD.map(lambda x: (x[0], x[1]))

复制代码

3 模型练习

基于ALS的原理，需要确认最佳秩（rank）值，起首循环计算多个秩的值，并记载最小误差，误差评价标准是RMSE，以最佳秩值作为输入重新举行练习，生成模型，代码如下：

1. #ALS参数配置
2. seed = 5
3. iterations = 10
4. regularization_param = 0.1
5. ranks = [4, 8, 12]
6. errors = [0, 0, 0]
7. err = 0
8. tolerance = 0.02
10. #模型训练确认rank值（最小误差）
11. min_error = float('inf')
12. best_rank = -1
13. best_iteration = -1
14. for rank in ranks:
15.     model = ALS.train(training_RDD, rank, seed=seed, iterations=iterations, lambda_=regularization_param)
16.     predict = model.predictAll(validation_predict_RDD).map(lambda r: ((r[0], r[1]), r[2]))
17.     rates_predictions = validation_RDD.map(lambda r: ((int(r[0]), int(r[1])), float(r[2]))).join(predict)
18.     error = math.sqrt(rates_predictions.map(lambda r: (r[1][0] - r[1][1]) ** 2).mean())
19.     errors[err] = error
20.     err += 1
21.     if error < min_error:
22.         min_error = error
23.         best_rank = rank
25. #以最佳rank值新重训练模型
26. model = ALS.train(training_RDD, best_rank, seed=seed, iterations=iterations, lambda_=regularization_param)

复制代码

练习好的模型可以生存到文件中，如许在使用模型时，通过矩阵分解模型（MatrixFactorization Model）从文件中加载即可使用，代码如下：

1. model.save(sc, "spark_movie.model")
2. sameModel = MatrixFactorizationModel.load(sc, "spark_movie.model")

复制代码

4 模型结果评估

模型的评估是将测试集（只含有用户编号和电影编号）提交给模型，由其举行打分，并将打分结果与测试会集实际评分值举行比较，对所有预测结果计算RMSE值，作为模型评价指标，代码如下，其输出结果为94.15%，阐明模型预测的准确性较好。

1. #模型测试
2. predictions = model.predictAll(test_predict_RDD).map(lambda r: ((r[0], r[1]), r[2]))
3. rates_and_predictions = test_RDD.map(lambda r: ((int(r[0]), int(r[1])), float(r[2]))).join(predictions)
4. #计算RMSE指标
5. error = math.sqrt(rates_predictions.map(lambda r: (r[1][0] - r[1][1]) ** 2).mean())
6. print('Model RMSE = %s' % error)

复制代码

5 模型使用

假设用户编号为16，如果要预测其对电影编号为48的电影的评分值，可通过以下代码，调用模型的predict方法即可返回评分值。别的，
还可以基于用户的汗青行为向其推荐n部电影，具体调用模型的recommendProducts方法，输入参数为用户编号和预测的电影数量，本实行中推荐电影的数量为10。

1. #预测某一用户对某一电影的评分
2. user_id = 16
3. movie_id = 48
4. predictedRating = model.predict(user_id, movie_id)
5. print("用户编号:"+str(user_id)+" 对电影:"+str(movie_id)+" 的评分为:"+str(predictedRating))
7. #向某一用户推荐10部电影
8. topKRecs = model.recommendProducts(user_id, 10)
9. print("向用户编号:"+str(user_id)+"的用户推荐10部电影:")
10. for rec in topKRecs:
11.     print(rec)

复制代码

运行程序后的输出结果如下，可以看到模型预测编号为16的用户对电影48评分为3.10。也可以看到向其推荐的10部电影编号和对应的评分 值。结果如下：

1.         Model RMSE = 0.9403545117603078
2.         用户编号:16 对电影:48 的评分为:3.216689261395394
3.         向用户编号:16的用户推荐10部电影:
4.         Rating(user=16, product=4630, rating=6.05529107837512)
5.         Rating(user=16, product=5765, rating=6.05529107837512)
6.         Rating(user=16, product=3437, rating=6.05529107837512)
7.         Rating(user=16, product=9010, rating=6.05529107837512)
8.         Rating(user=16, product=7371, rating=5.807446390235324)
9.         Rating(user=16, product=8607, rating=5.802511774802554)
10.         Rating(user=16, product=91653, rating=5.751695420024291)
11.         Rating(user=16, product=2810, rating=5.648375121054762)
12.         Rating(user=16, product=6063, rating=5.630159072112605)
13.         Rating(user=16, product=7247, rating=5.582917190407091)

复制代码

表现电影标题

• header： csv文件的header。默认值是false；
  
• delimiter： 分隔符。默认值是’ , '；
  
• inferSchema： 根据你的数据预测你的数据范例，加了的话读取的次数是2次。这么说吧，比如学生的成绩，你不加的话，读出来的范例是string，加了就是int。
  

1. sqlsc = SQLContext(sc)
2. movieTitle = sqlsc.read.options(header='true', inferSchema='True', delimiter=',').csv("movies.csv")
3. print("向用户编号:"+str(user_id)+"的用户推荐10部电影:")
5. for p in topKRecs:
6.     print(movieTitle.filter("movieId=="+str(p[1])).show())  # <class 'pyspark.sql.dataframe.DataFrame

复制代码

打印结果：

1.         向用户编号:16的用户推荐10部电影:
2.         +-------+--------------------+------+
3.         |movieId|               title|genres|
4.         +-------+--------------------+------+
5.         |   4630|No Holds Barred (...|Action|
6.         +-------+--------------------+------+
7.        
8.         None
9.         +-------+-----+------+
10.         |movieId|title|genres|
11.         +-------+-----+------+
12.         +-------+-----+------+
13.        
14.         None
15.         +-------+-----+------+
16.         |movieId|title|genres|
17.         +-------+-----+------+
18.         +-------+-----+------+
19.        
20.         None
21.         +-------+--------------------+-------------+
22.         |movieId|               title|       genres|
23.         +-------+--------------------+-------------+
24.         |   9010|Love Me If You Da...|Drama|Romance|
25.         +-------+--------------------+-------------+
26.        
27.         None
28.         +-------+---------------+--------------------+
29.         |movieId|          title|              genres|
30.         +-------+---------------+--------------------+
31.         |   7371|Dogville (2003)|Drama|Mystery|Thr...|
32.         +-------+---------------+--------------------+
33.        
34.         None
35.         +-------+--------------------+--------------------+
36.         |movieId|               title|              genres|
37.         +-------+--------------------+--------------------+
38.         |   8607|Tokyo Godfathers ...|Adventure|Animati...|
39.         +-------+--------------------+--------------------+
40.        
41.         None
42.         +-------+--------------------+------------+
43.         |movieId|               title|      genres|
44.         +-------+--------------------+------------+
45.         |  91653|We Bought a Zoo (...|Comedy|Drama|
46.         +-------+--------------------+------------+
47.        
48.         None
49.         +-------+-------------------+--------------------+
50.         |movieId|              title|              genres|
51.         +-------+-------------------+--------------------+
52.         |   2810|Perfect Blue (1997)|Animation|Horror|...|
53.         +-------+-------------------+--------------------+
54.        
55.         None
56.         +-------+----------+------------+
57.         |movieId|     title|      genres|
58.         +-------+----------+------------+
59.         |   6063|May (2002)|Drama|Horror|
60.         +-------+----------+------------+
61.        
62.         None
63.         +-------+--------------------+--------------------+
64.         |movieId|               title|              genres|
65.         +-------+--------------------+--------------------+
66.         |   7247|Chitty Chitty Ban...|Adventure|Childre...|
67.         +-------+--------------------+--------------------+
68.        
69.         None

复制代码

针对电影4754推荐最有可能喜欢的前5个用户：

1. model.recommendUsers(4754, 5)

复制代码

输出结果：

1. [Rating(user=156, product=4754, rating=7.392522617118351),
2. Rating(user=304, product=4754, rating=6.951528110514651),
3. Rating(user=498, product=4754, rating=6.902421770828882),
4. Rating(user=348, product=4754, rating=6.877413192449223),
5. Rating(user=29, product=4754, rating=6.870045800604159)]

复制代码

八、实训总结

通过本实行的结果可以看到，基于Spark的协同过滤推荐算法并没有依赖具体的业务数据，如电影的内容分析和用户属性特性分析等，阐明它为通用算法框架，可以用于其他行业的个性化推荐，比方餐饮推荐、音乐推荐和新闻推荐等，只要将各行业中的评分数据转化为本例中的ratings.csv对应的格式即可直策应用。
九、附资源

开头处无需积分即可下载