每天天都在努力学习的我们
前言
本篇博客讲解的内容依旧是使用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项目就是离线数仓项目的建设。文章来源:https://www.toymoban.com/news/detail-482369.html
最后,希望我的她可以越来越好,天天开心文章来源地址https://www.toymoban.com/news/detail-482369.html
到了这里,关于基于Spark的音乐专辑数据分析的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!