【版权声明】版权所有，严禁转载，严禁用于商业用途，侵权必究。作者：厦门大学信息学院计算机科学系2018级研究生 沈琳指导老师：厦门大学数据库实验室 林子雨 博士/副教授相关教材：林子雨、郑海山、赖永炫编著《Spark编程基础（Python版）》（访问教材官网）相关案例：基于Python语言的Spark数据处理分析案例集锦（PySpark）本实验采用Python语言，使用大数据处理框架Spark对音乐专辑数据进行处理分析，并对分析结果进行可视化。一、实验环境（1）Linux： Ubuntu 16.04（2）Python: 3.6（3）Hadoop：3.1.3（查看安装教程）（4）Spark: 2.4.0 （查看安装教程）（5）Web框架：flask 1.0.3（6）可视化工具：Echarts（7）开发工具：Visual Studio Code安装完上述环境以后，为了支持Python可视化分析，还需要执行如下命令安装Flask组件：sudo apt-get install python3-pip pip3 install flask 本实验涉及到的所有数据集和代码，可以从百度网盘下载（下载地址，提取码：jp0j）。二、实验数据集1.数据集说明在Kaggle数据平台上下载了数据集albums.csv。其中包含了10万条音乐专辑的数据。主要字段说明如下：1) album_title：音乐专辑名称2) genre：专辑类型3) year_of_pub： 专辑发行年份4) num_of_tracks： 每张专辑中单曲数量5) num_of_sales：专辑销量6) rolling_stone_critic：滚石网站的评分7) mtv_critic：全球最大音乐电视网MTV的评分8) music_maniac_critic：音乐达人的评分2.将数据集存放在分布式文件系统HDFS中A. 启动Hadoop中的HDFS组件，在命令行运行下面命令：/usr/local/hadoop/sbin/start-dfs.sh B. 在hadoop上登录用户创建目录，在命令行运行下面命令：hdfs dfs -mkdir -p /user/hadoop C. 把本地文件系统中的数据集albums.csv上传到分布式文件系统HDFS中：hdfs dfs -put albums.csv 三、pyspark的简单分析1.建立工程文件A. 创建文件夹code。B. 在code下创建main.py文件。C. 在code下创建static文件夹，存放静态文件。D. 在code/static文件夹下面创建data目录，存放分析生成的json数据。2.进行数据分析本文对音乐专辑数据集albums.csv进行了一系列的分析，包括：1) 统计各类型专辑的数量；2) 统计各类型专辑的销量总数；3) 统计近20年每年发行的专辑数量和单曲数量；4) 分析总销量前五的专辑类型的各年份销量；5) 分析总销量前五的专辑类型，在不同评分体系中的平均评分。3.具体代码在main.py中复制以下代码：from pyspark import SparkContext from pyspark.sql import SparkSession import json #统计各类型专辑的数量（只显示总数量大于2000的十种专辑类型） def genre(sc, spark, df): #按照genre字段统计每个类型的专辑总数，过滤出其中数量大于2000的记录 #并取出10种类型用于显示 j = df.groupBy('genre').count().filter('count > 2000').take(10) #把list数据转换成json字符串，并写入到static/data目录下的json文件中 f = open('static/data/genre.json', 'w') f.write(json.dumps(j)) f.close() #统计各个类型专辑的销量总数 def genreSales(sc, spark, df): j = df.select('genre', 'num_of_sales').rdd\ .map(lambda v: (v.genre, int(v.num_of_sales)))\ .reduceByKey(lambda x, y: x + y).collect() f = open('static/data/genre-sales.json', 'w') f.write(json.dumps(j)) f.close() #统计每年发行的专辑数量和单曲数量 def yearTracksAndSales(sc, spark, df): #把相同年份的专辑数和单曲数量相加，并按照年份排序 result = df.select('year_of_pub', 'num_of_tracks').rdd\ .map(lambda v: (int(v.year_of_pub), [int(v.num_of_tracks), 1]))\ .reduceByKey(lambda x, y: [x[0] + y[0], x[1] + y[1]])\ .sortByKey()\ .collect() #为了方便可视化实现，将列表中的每一个字段分别存储 ans = {} ans['years'] = list(map(lambda v: v[0], result)) ans['tracks'] = list(map(lambda v: v[1][0], result)) ans['albums'] = list(map(lambda v: v[1][1], result)) f = open('static/data/year-tracks-and-sales.json', 'w') f.write(json.dumps(ans)) f.close() #取出总销量排名前五的专辑类型 def GenreList(sc, spark, df): genre_list = df.groupBy('genre').count()\ .orderBy('count',ascending = False).rdd.map(lambda v: v.genre).take(5) return genre_list #分析总销量前五的类型的专辑各年份销量 def GenreYearSales(sc, spark, df, genre_list): #过滤出类型为总销量前五的专辑，将相同类型、相同年份的专辑的销量相加，并进行排序。 result = df.select('genre', 'year_of_pub', 'num_of_sales').rdd\ .filter(lambda v: v.genre in genre_list)\ .map(lambda v: ((v.genre, int(v.year_of_pub)), int(v.num_of_sales)))\ .reduceByKey(lambda x, y: x + y)\ .sortByKey().collect() #为了方便可视化数据提取，将数据存储为适配可视化的格式 result = list(map(lambda v: [v[0][0], v[0][1], v[1]], result)) ans = {} for genre in genre_list: ans[genre] = list(filter(lambda v: v[0] == genre, result)) f = open('static/data/genre-year-sales.json', 'w') f.write(json.dumps(ans)) f.close() #总销量前五的专辑类型，在不同评分体系中的平均评分 def GenreCritic(sc, spark, df, genre_list): #过滤出类型为总销量前五的专辑，将同样类型的专辑的滚石评分、mtv评分，音乐达人评分分别取平均 result = df.select('genre', 'rolling_stone_critic', 'mtv_critic', 'music_maniac_critic').rdd\ .filter(lambda v: v.genre in genre_list)\ .map(lambda v: (v.genre, (float(v.rolling_stone_critic), float(v.mtv_critic), float(v.music_maniac_critic), 1)))\ .reduceByKey(lambda x, y : (x[0] + y[0], x[1] + y[1], x[2] + y[2], x[3] + y[3]))\ .map(lambda v: (v[0], v[1][0]/v[1][3], v[1][1]/v[1][3], v[1][2]/v[1][3])).collect() f = open('static/data/genre-critic.json', 'w') f.write(json.dumps(result)) f.close() #代码入口 if __name__ == "__main__": sc = SparkContext( 'local', 'test') sc.setLogLevel("WARN") spark = SparkSession.builder.getOrCreate() file = "albums.csv" df = spark.read.csv(file, header=True) #dataframe genre_list = GenreList(sc, spark, df) genre(sc, spark, df) genreSales(sc, spark, df) yearTracksAndSales(sc, spark, df) GenreYearSales(sc, spark, df, genre_list) GenreCritic(sc, spark, df, genre_list) 由于代码中已有详细注释，在此不多加解释。4.代码运行A. 在Ubuntu终端窗口中，用 hadoop 用户登录，在命令行运行su hadoop，并输入用户密码。B. 进入代码所在目录。C. 为了能够读取HDFS中的albums.csv文件，在命令行运行：/usr/local/hadoop/sbin/start-dfs.sh D. 在命令行运行：spark-submit main.py 四、可视化实现本实验的可视化基于Echarts实现，实现的可视化页面部署在基于flask框架的web服务器上。1. 相关代码结构1) 在code目录下新建SparkFlask.py文件，存放Flask 应用。2) 在code目录下新建一个名为templates的文件夹，存放html文件。3) 在code/static目录下新建一个名为js的文件夹，存放js文件。最后的代码结构如下图所示：2. 建立Flask应用在SparkFlask.py文件中复制以下代码：from flask import render_template from flask import Flask # from livereload import Server app = Flask(__name__) @app.route('/') def index(): #使用 render_template() 方法来渲染模板 return render_template('index.html') @app.route('/') def req_file(filename): return render_template(filename) if __name__ == '__main__': app.DEBUG=True#代码调试立即生效 app.jinja_env.auto_reload = True#模板调试立即生效 app.run()#用 run() 函数来让应用运行在本地服务器上 代码中已有详细注释，在此不多加解释。3.下载js文件1) 在网站上下载jQuery（下载），将其另存为jquery.min.js文件，保存在code/static/js目录下。2) 在官网下载界面下载Echarts（下载），将其另存echarts-gl.min.js文件，保存在code/static/js目录下。4.Echarts画图（1）在code/templates目录下新建index.html文件。复制以下代码：

音乐专辑分析

• 各类型专辑的数量统计图
• 各类型专辑的销量统计图
• 近20年每年发行的专辑数量和单曲数量统计图
• 总销量前五的专辑类型的各年份销量分析图
• 总销量前五的专辑类型的评分分析图

index.html为主页面，显示每一个统计分析图所在页面的链接。点击任意一个链接，即可跳转到相应页面。（2）在code/templates目录下新建genre.html文件。复制以下代码： Return

这个通过读取code/static/data/genre.json中的数据，画出玫瑰图，显示各类型专辑的数量。（3）在code/templates目录下新建genre-sales.html文件。复制以下代码： Return

这个通过读取code/static/data/genre-sales.json中的数据，画出柱状图，显示各类型专辑的销量总数。（4）在code/templates目录下新建year-tracks-and-sales.html文件。复制以下代码： Return

这个通过读取code/static/data/ year-tracks-and-sales.json中的数据，画出柱状图，显示近20年每年发行的专辑数量和单曲数量。（5）在code/templates目录下新建genre-year-sales.html文件。复制以下代码： Return

这个通过读取code/static/data/ genre-year-sales.json中的数据，画出扇形图和折线图，分别显示不同年份各类型专辑的销量占总销量的比例，和总销量前五的专辑类型的各年份销量变化。（6）在code/templates目录下新建genre-critic.html文件。复制以下代码： Return

这个通过读取code/static/data/ genre-critic.json中的数据，画出柱形图，显示总销量前五的专辑类型，在不同评分体系中的平均评分。5. 启动程序A. 在另一个Ubuntu终端窗口中，用 hadoop 用户登录，在命令行运行su hadoop，并输入用户密码。B. 进入代码所在目录。C. 在命令行运行如下命令：spark-submit SparkFlask.py D. 在浏览器打开http://127.0.0.1:5000/，可看到如下界面：1) 各类型专辑的数量统计图2) 各类型专辑的销量统计图3) 近20年每年发行的专辑数量和单曲数量统计图4) 总销量前五的专辑类型的各年份销量分析图5) 总销量前五的专辑类型的评分分析图