Apache Spark 练习五使用Spark进行YouTube视频网站指标

一、源数据

本章所分析的数据来自于Simon Fraser大学公开的YouTube视频网站的视频数据（https://netsg.cs.sfu.ca/youtubedata/）。数据包含两张表。第一张为视频表，记录了研究人员爬取的视频的元数据信息，具体包括以下字段：

第二张表为用户表，记录了爬取的YouTube用户的相关信息，具体包括：

二、练习题

0. 数据预处理

本章所分析的视频信息下载自http://netsg.cs.sfu.ca/youtubedata/080327.zip，我们将该压缩包中的所有文件进行了归并，并过滤掉那些字段数不足10个的记录。此外，我们将category字段的数据进行了预处理，将所有的类别用&分割，同时去掉两边空格。并且，多个相关视频id也使用&进行分割。用户信息则下载自https://netsg.cs.sfu.ca/youtubedata/080903user.zip。然后，我们将这些数据读取为Spark DataFrame形式，以供后续分析。

1.  
   val spark = SparkSession
2.  
   .builder()
3.  
   .appName("Youtube")
4.  
   .getOrCreate()
5.  
   import spark.implicits._
6.  
    
7.  
   /* 加载源数据 */
8.  
   // 源数据下载自 https://netsg.cs.sfu.ca/youtubedata/
9.  
   // 加载视频数据
10.  
    val videoRDD =
11.  
    spark.sparkContext.textFile("hdfs:///SparkLearning/youtube_video.txt")
12.  
    val videoSchema = StructType(
13.  
    Array[StructField](
14.  
    StructField("video_id", StringType, nullable = true),
15.  
    StructField("uploader", StringType, nullable = true),
16.  
    StructField("age", IntegerType, nullable = true),
17.  
    StructField("category", ArrayType(StringType), nullable = true),
18.  
    StructField("length", IntegerType, nullable = true),
19.  
    StructField("views", IntegerType, nullable = true),
20.  
    StructField("rate", DoubleType, nullable = true),
21.  
    StructField("ratings", IntegerType, nullable = true),
22.  
    StructField("comments", IntegerType, nullable = true),
23.  
    StructField("related_ids", ArrayType(StringType), nullable = true)
24.  
    )
25.  
    )
26.  
    val rowVideoRDD = videoRDD
27.  
    .map(_.split("\t"))
28.  
    .map(attributes =>
29.  
    Row(
30.  
    attributes(0),
31.  
    attributes(1),
32.  
    attributes(2).toInt,
33.  
    attributes(3).split("&"),
34.  
    attributes(4).toInt,
35.  
    attributes(5).toInt,
36.  
    attributes(6).toDouble,
37.  
    attributes(7).toInt,
38.  
    attributes(8).toInt,
39.  
    attributes(9).split("&")
40.  
    )
41.  
    )
42.  
    val videoDF = spark.createDataFrame(rowVideoRDD, videoSchema)
43.  
     
44.  
    // 加载用户数据
45.  
    val userRDD =
46.  
    spark.sparkContext.textFile("hdfs:///SparkLearning/youtube_user.txt")
47.  
    val userSchema = StructType(
48.  
    Array[StructField](
49.  
    StructField("uploader", StringType, nullable = true),
50.  
    StructField("videos", IntegerType, nullable = true),
51.  
    StructField("friends", IntegerType, nullable = true)
52.  
    )
53.  
    )
54.  
    val rowUserRDD = userRDD
55.  
    .map(_.split("\t"))
56.  
    .map(attributes =>
57.  
    Row(attributes(0), attributes(1).toInt, attributes(2).toInt)
58.  
    )
59.  
    val userDF = spark.createDataFrame(rowUserRDD, userSchema)

1. 统计视频观看数Top10

1.  
   val res = videoDF
2.  
   .select($"video_id", $"views")
3.  
   .orderBy($"views".desc)
4.  
   .limit(10)

2. 统计视频类别热度Top10

1.  
   val res = videoDF
2.  
   .select(explode($"category").as("category"))
3.  
   .groupBy($"category")
4.  
   .count()
5.  
   .orderBy($"count".desc)
6.  
   .limit(10)

3. 统计出视频观看数最高的20个视频的所属类别以及类别包含Top20视频的个数

1.  
   val res = videoDF
2.  
   .orderBy($"views".desc)
3.  
   .limit(20)
4.  
   .select(explode($"category").as("category"))
5.  
   .groupBy($"category")
6.  
   .count()

4. 统计视频观看数Top50所关联视频的所属类别Rank

1.  
   val res = videoDF
2.  
   .orderBy($"views".desc)
3.  
   .limit(50)
4.  
   .select(explode($"related_ids").as("related_id"))
5.  
   .alias("t1")
6.  
   .join(videoDF.as("t2"), $"t1.related_id" === $"t2.video_id")
7.  
   .select(explode($"t2.category").as("category"))
8.  
   .groupBy($"category")
9.  
   .count()
10.  
    .orderBy($"count".desc)

5. 统计每个类别中的视频观看数Top10

1.  
   val res = videoDF
2.  
   .select($"video_id", explode($"category").as("category"), $"views")
3.  
   .select(
4.  
   $"category",
5.  
   $"video_id",
6.  
   $"views",
7.  
   row_number()
8.  
   .over(Window.partitionBy($"category").orderBy($"views".desc))
9.  
   .alias("rank")
10.  
    )
11.  
    .filter($"rank" <= 10)
12.  
    .orderBy($"category", $"rank")

6. 统计上传视频最多的用户Top10以及他们上传的视频

1.  
   val res = userDF
2.  
   .orderBy($"videos".desc)
3.  
   .limit(10)
4.  
   .alias("t1")
5.  
   .join(videoDF.alias("t2"), $"t1.uploader" === $"t2.uploader")
6.  
   .select($"t2.uploader", $"t2.video_id")
7.  
   .orderBy($"uploader")

这篇好文章是转载于：学新通技术网