一、源数据
本章所分析的数据来自于Kaggle公开的、人工合成的音乐专辑发行数据（https://www.kaggle.com/datasets/revilrosa/music-label-dataset）。以下，我们只针对albums.csv文件进行分析。该数据具体包括以下字段：
- id: the album identifier;
- artist_id: the artist identifier;
- album_title: the title of the album;
- genre: the genre of the album. An artist can release albums of different genres;
- year_of_pub: the year the album was published;
- num_of_tracks: how many tracks there are in the album (a small number can mean longer tracks);
- num_of_sales: how many sales the album has made in the first month after the release;
- rolling_stone_critic: how magazine Rolling Stone has rated the album;
- mtv_critic: how MTV has rated the album;
- music_maniac_critic: how review site Music Maniac has rated the album.
二、练习题
0. 数据预处理
以下，我们将csv文件从HDFS中读取进来，并转换为Spark DataFrame格式。
val spark = SparkSession
.builder()
.appName("Albums")
.getOrCreate()
import spark.implicits._
val df = spark.read
.option("header", "true")
.option("inferSchema", "true")
.csv("hdfs:///SparkLearning/albums.csv")
1. 统计各类型专辑的数量
val res = df
.select($"genre")
.groupBy($"genre")
.count()
.orderBy($"count".desc)
2. 统计各类型专辑的销量总数
val res = df
.select($"genre", $"num_of_sales")
.groupBy($"genre")
.sum("num_of_sales")
.withColumnRenamed("sum(num_of_sales)", "total_sales")
.orderBy($"total_sales".desc)
3. 统计近20年每年发行的专辑数量和单曲数量
val res = df
.select("year_of_pub", "num_of_tracks")
.filter($"year_of_pub" >= 2000)
.groupBy($"year_of_pub")
.agg("num_of_tracks" -> "count", "num_of_tracks" -> "sum")
.withColumnRenamed("count(num_of_tracks)", "total_albums")
.withColumnRenamed("sum(num_of_tracks)", "total_tracks")
.orderBy("year_of_pub")
4. 分析总销量前五的专辑类型的各年份销量
val res = df
.select($"genre", $"num_of_sales")
.groupBy("genre")
.sum("num_of_sales")
.withColumnRenamed("sum(num_of_sales)", "total_sales")
.orderBy($"total_sales".desc)
.limit(5)
.alias("t1")
.join(
df.select($"genre", $"num_of_sales", $"year_of_pub").alias("t2"),
$"t1.genre" === $"t2.genre"
)
.groupBy("t2.genre", "t2.year_of_pub")
.sum("t2.num_of_sales")
.orderBy($"genre", $"year_of_pub")
5. 分析总销量前五的专辑类型，在不同评分体系中的平均评分
val res = df
.select($"genre", $"num_of_sales")
.groupBy("genre")
.sum("num_of_sales")
.withColumnRenamed("sum(num_of_sales)", "total_sales")
.orderBy($"total_sales".desc)
.limit(5)
.alias("t1")
.join(
df.select(
$"genre",
$"rolling_stone_critic",
$"mtv_critic",
$"music_maniac_critic"
).alias("t2"),
$"t1.genre" === $"t2.genre"
)
.groupBy("t2.genre")
.agg(
"rolling_stone_critic" -> "avg",
"mtv_critic" -> "avg",
"music_maniac_critic" -> "avg"
)
.orderBy($"genre")
