基于Spark的音乐专辑数据分析

每天天都在努力学习的我们

前言

本篇博客讲解的内容依旧是使用Spark进行相关的数据分析，按理来说数据分析完之后应该搞一搞可视化的，由于目前时间紧张，顾不得学习可视化了，先来看一下此次的内容把。

在Kaggle数据平台下载了数据集albunms.csv，里面包含了的主要字段如下，先来看一下。

使用Spark读取csv

spark读取csv的方式有两种，一种是使用rdd进行读取csv,然后创建RDD对象。另一种是使用spark SQL进行读取，创建DataFrame对象。本篇博客使用Spark SQL进行读取文件，RDD和Data Frame处理数据。

csv文件我们都知道，","分隔符，但是读取csv文件的同时也要注意是否有无表头，表头字段类型。下面来看一下Spark SQL读取csv。

private
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark SQL") private
val sparkContext = new SparkContext(sparkConf)
private val sparkSession: SparkSession =SparkSession.builder().config(sparkConf).getOrCreate() import sparkSession.implicits._ def
Transform_demo()={
//使用sparksession 读取csv文件，此csv文件有表头
val dataFrame = sparkSession.read.format("com.databricks.spark.csv")
//有无表头
.option("header", true)//是否自动推断表头类型.option("inferSchema", false)
//分隔符
.option("delimiter", ",")
//csv文件的地址
.csv("date/albums.csv")
dataFrame
}

统计各类型专辑的数量

思路：

各类型专辑的数量 ==》根据专辑类型(genre)分组，求出专辑类型的总和

使用spark SQL进行处理

def genre_demo() = {//统计各个类型专辑的数量
val unit = Transform_demo()
.select($"genre")
.groupBy($"genre")
.count()
.sort($"count".desc)
.show() }

在Spark SQL中使用group by的时候，提供了group by之后的操作，比如

max(colNames:String*):获取分组中指定字段或者所有的数字类型字段的最大值，只能作用于数字型字段。

min(colNames:String*):获取分组字段或者所有的数字类型字段的最小值，只能作用于数字类型的字段。

mean(colName:String*):获取分组中指定字段或者所有数字类型字段的平均值，只能作用于数据类型的字段

sum(colNames:String*)：获取分组中指定字段或者所有数字类型字段的累加值，只能作用于数字类型的字段

count():获取分组中的元素个数

根据genre字段分组后，求出专辑类型的个数，对专辑类型的个数进行排序。如果需要导入到文件里面，那么需要.write。

使用Rdd进行处理

     def genre_demo() = {val value = Transform_demo()
.select($"genre")
.rdd.map(v => (v(0), 1)).reduceByKey(_ + _).sortBy(_._2,false)

展示一下上面的流程图把

统计各类型专辑的销量总数

思路：

根据专辑类型分组，分组之后，计算num_of_sales专辑销量总和。

如果是spark SQL的话，流程应该是这样的。

dataFrame => select => group by => sum()

来看一下代码

def countByNum_sales()={Transform_demo()
.select($"genre",$"num_of_sales")
.withColumn("num_of_sales",col("num_of_sales").cast("Integer"))
.groupBy("genre").sum("num_of_sales")
.orderBy($"sum(num_of_sales)".desc).show() }

如果在group by后面直接sum求和，那么是会报错的，因为在最开始的时候，我们并没有让系统自动推断表头的数据类型，默认为String类型，因此需要先转换为整型，然后对其进行操作。

来看一下RDD的代码

def countByNum_sales()={val value = Transform_demo()
.select($"genre", $"num_of_sales")
.rdd
.map(v => (v(0).toString, v(1).toString.toInt))
.reduceByKey(_ + _) .sortBy(_._2, false)
value }

统计近20年每年发行的专辑数量和单曲数量

思路：

根据年份分组(year_of_pub),求每年发行的专辑数量和单曲数量。

单曲数量很简单(num_of_tracks)累加，但是专辑数量怎么表示呢？

注意一下，num_of_tracks字段：每张专辑中的单曲数量，什么意思呢？来看一下这样的表示

(num_of_tracks，1) =》 (每张专辑的单曲数量，专辑数量)

这个1就代表着专辑的数量。

用RDD来理一下思路，(年份,(每张专辑的单曲数量，1)) ==>经过reduceByKey (年份,(每年发行的单曲数量，每年发行的专辑数量))

def Countbytracks() ={//统计近20年每年发行的专辑数量和单曲数量；//每年发行的单曲数量 num_of_tracks 专辑数量怎么算呢？//num_of_tracks： 每张专辑中单曲数量//(1张专辑，专辑发行的单曲数量)
val array = Transform_demo()
.select($"year_of_pub", $"num_of_tracks")
.rdd
.map(v => (v(0).toString.toInt, (v(1).toString.toInt, 1))).reduceByKey((x, y) => (x._1 + y._1, (x._2 + y._2))).sortByKey()
array }

来看一下流程图

分析总销量前五的专辑类型的各年份销量

思路：

首先求出来总销售量前五的专辑类型，然后取前五个数量最多的，之后在这个五个数量最多的专辑类型里面计算每年的销量。

首先获取总销售量前五的专辑类型

def get_genre() = {
//先获取总销售量前5的专辑类型
val array = Transform_demo()
.select($"genre", $"num_of_sales")
.withColumn("num_of_sales", col("num_of_sales").cast("Integer"))
.groupBy($"genre").sum("num_of_sales").orderBy($"sum(num_of_sales)".desc)
.rdd
.map(v => v(0).toString)
.take(5)
array }

首先更改字段类型，根据销售类型(genre)分组，找到销量最多的销售类型.经过map转换之后取前五个销售类型。

获取总销量前5的专辑类型的各年份销量

def per_year_sales()={val genre_list = get_genre()val value = Transform_demo()
.select($"genre", $"year_of_pub", $"num_of_sales").rdd.filter(v => genre_list.contains(v(0)) ).map(v => ((v(0).toString, v(1).toString), v(2).toString.toInt))
.reduceByKey(_ + _)value }

如果单独看上面的代码看不懂，那么来看一下下面的流程图把，一定明明白白。

总结

关于这篇音乐分析项目就到此为止了。

关于对数据的基本处理已经有了些眉目，接下来就是不断的练习练习。

之后还有会其它的项目，我们最终的Boss项目就是离线数仓项目的建设。

最后，希望我的她可以越来越好，天天开心