Spark上机实验手册

1、实验六：Spark MLLib实验 【实验目的】使用SparkMLLib中的推荐算法接口，实现电影推荐，对使用Spark进行推荐有 个直观的认识【实验原理】1、 使用协同过滤算法训练模型2、 调用模型根据用户id和电影推荐的数量进行推荐【实验环境】操作系统：Linux开发环境：pyspark交互式命令行wget :8888/zz/Files/u.userwget :8888/zz/Files/u.datawget :8888/zz/Files/u.itemmv u.user /root/filmmv

2、 u.data /root/filmmv u.item /root/filmpip install numpy -i /simple【实验步骤】MovieLens数据集包含多个用户对多部电影的评级数据，也包括电影元数据信息和用户属性信 息。这个数据集经常用来做推荐系统，机器学习算法的测试数据集。尤其在推荐系统领域，很多 著名论文都是基于这个数据集的。下载地址为：http :/files.group /datasets/movielens/， 本文下 载数据 100k版本，对该数据集进行探索：下载后解压文件，本实验使

3、用其中三个文件，用户信息、影片信息、评分数据。用户信息（u.user）影片信息(u.item)评分数据(u.data)一、 代码实现#读取用户数据user_df=spark.read.text(file:/root/film/u.user)user_df.show(10)#为用户数据添加 schemafrom pyspark import Rowuser_rdd=user_df.rdd.map(lambda x:x0.split(|).map(lambda x:Row(id=x0,age=x1,sex=x2,job=x3,code=x4)user_rdd.take(5)#创建用户 datafr

4、ameuser_df=spark.createDataFrame(user_rdd)#注册临时表user_df.registerTempTable(user)spark.sql(select id,age,sex,job,code from user).show()#读取评分数据rating_df=spark.read.text(file:/root/film/u.data)rating_df.show(5)#为评分数据添加 schemarating_rdd=rating_df.rdd.map(lambda x:x0.split().map(lambda x:Row(user_id=x0,fi

5、lm_id=x1,rating=x2,time=x3)rating_rdd.take(5)#创建评分 dataframerating_df=spark.createDataFrame(rating_rdd)#注册评分临时表rating_df.registerTempTable(rating)spark.sql(select user_id,film_id,rating,time from rating).show()#读取电影信息film_df=spark.read.text(file:/root/film/u.item)film_df.show(5)#为电影数据添加 schemafilm_r

6、dd=film_df.rdd.map(lambda x:x0.split(|).map(lambda x:Row(id=x0,title=x1,post_time=x2,url=x4)film_rdd.take(5)#创建电影 dataframefilm_df=spark.createDataFrame(film_rdd)#注册电影临时表film_df.registerTempTable(film)spark.sql(select * from film).show()#导入 Spark 机器学习包： 协同过滤算法(需要之前安装好numpy)from pyspark.mllib.recomme

7、ndation import ALS#ALS.trainImplicit 中第一个参数 RDD 是固定格式（ 用户编号， 电影编号， 评分值）rdd=rating_rdd.map(lambda x:(x3,x0,x1)rdd.take(5)#训练模型train参数见/weixin_40717296/article/details/79695694model=ALS.train(rdd,10,10,0.01)#使用模型进行推荐， 参数为用户 id， 推荐个数user_film=model.recommendProducts(112,3)print(user

8、_film)#显示推荐电影名称type(user_film)for film in user_film: spark.sql(select id ,title,post_time,url , +str(film2)+ as similar from film where id=+str(film1).show()#查看用户看过的电影spark.sql(select a.user_id,b.title from rating a,film b where a.film_id=b.id and a.user_id=112).show()#保存模型import osimport shutilsave_path=/root/film/modelif os.path.exists(save_path): shutil.rmtree(save_path)model.save(sc,save_path)print(model saved!)#调用模型from pyspark.mllib.recommendation import MatrixFactorizationModelmodel=MatrixFactorizationModel.load(sc,save_path)result=model.recommendPr