别再人云亦云了！！！你真的搞懂了RDD、DataFrame、DataSet的区别吗？

1
2
3
4
5
package object sql {
  // ...
  type DataFrame = Dataset[Row]
}

1
2
https://www.kaggle.com/datasnaek/youtube-new?select=USvideos.csv

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <scala.version>2.12</scala.version>
        <spark.version>3.0.1</spark.version>
    </properties>

    <repositories>
        <repository>
            <id>central</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
        </repository>
    </repositories>
    <pluginRepositories>
        <pluginRepository>
            <id>central</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
        </pluginRepository>
    </pluginRepositories>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>com.opencsv</groupId>
            <artifactId>opencsv</artifactId>
            <version>5.3</version>
        </dependency>
    </dependencies>

    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
    </build>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
/**
 * Spark RDD处理示例
 */
object RddAnalysis {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("RDD Process").setMaster("local[*]")
        val sc = new SparkContext(conf)

        // 读取本地文件创建RDD
        val youtubeVideosRDD = {
            sc.textFile("""E:5.git_projectdatasetyoutube""")
        }

        // 统计不同分类Youtube视频的喜欢人数、不喜欢人数
        // 1. 添加行号
        // 创建计数器
        val rownumAcc = sc.longAccumulator("rownum")
        // 带上行号
        youtubeVideosRDD.map(line => {
                rownumAcc.add(1)
                rownumAcc.value -> line
            })
            // 过滤掉第一行
            .filter(_._1 != 1)
            // 去除行号
            .map(_._2)
            // 过滤掉非法的数据
            .filter(line => {
                val fields = line.split("01")
                val try1 = scala.util.Try(fields(8).toLong)
                val try2 = scala.util.Try(fields(9).toLong)

                if(try1.isFailure || try2.isFailure)
                    false
                else
                    true
            })
            // 读取三个字段（视频分类、喜欢的人数、不喜欢的人数
            .map(line => {
                // 按照01解析CSV
                val fields = line.split("01")
                // 取第4个（分类）、第8个（喜欢人数）、第9个（不喜欢人数）
                // (分类id, 喜欢人数, 不喜欢人数)
                (fields(4), fields(8).toLong, fields(9).toLong)
            })
            // 按照分类id分组
            .groupBy(_._1)
            .map(t => {
                val result = t._2.reduce((r1, r2) => {
                    (r1._1, r1._2 + r2._2, r1._3 + r2._3)
                })
                result
            })
            .foreach(println)
    }
}

1
2
3
4
5
6
7
8
9
("BBC Three",8980120,149525)
("Ryan Canty",11715543,80544)
("Al Jazeera English",34427,411)
("FBE",9003314,191819)
("Sugar Pine 7",1399232,81062)
("Rob Scallon",11652652,704748)
("CamilaCabelloVEVO",19077166,1271494)
("Grist",3133,37)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
object DataFrameAnalysis {
    def main(args: Array[String]): Unit = {
        val spark = SparkSession
            .builder()
            .appName("Youtube Analysis")
            .master("local[*]")
            .config("spark.sql.shuffle.partitions",1)
            .getOrCreate()

        import spark.sqlContext.implicits._

        // 读取CSV
        val youtubeVideoDF = spark.read.option("header", true).csv("""E:5.git_projectdatasetUSvideos.csv""")

        import org.apache.spark.sql.functions._

        // 按照category_id分组聚合
        youtubeVideoDF.select($"category_id", $"likes".cast(LongType), $"dislikes".cast(LongType))
            .where($"likes".isNotNull)
            .where( $"dislikes".isNotNull)
            .groupBy($"category_id")
            .agg(sum("likes"), sum("dislikes"))
            .show()
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
== Parsed Logical Plan ==
GlobalLimit 1
+- LocalLimit 1
   +- Filter (length(trim(value#6, None)) > 0)
      +- Project [value#0 AS value#6]
         +- Project [value#0]
            +- Relation[value#0] text

== Analyzed Logical Plan ==
value: string
GlobalLimit 1
+- LocalLimit 1
   +- Filter (length(trim(value#6, None)) > 0)
      +- Project [value#0 AS value#6]
         +- Project [value#0]
            +- Relation[value#0] text

== Optimized Logical Plan ==
GlobalLimit 1
+- LocalLimit 1
   +- Filter (length(trim(value#0, None)) > 0)
      +- Relation[value#0] text

== Physical Plan ==
CollectLimit 1
+- *(1) Filter (length(trim(value#0, None)) > 0)
   +- FileScan text [value#0] Batched: false, DataFilters: [(length(trim(value#0, None)) > 0)], Format: Text, Location: InMemoryFileIndex[file:/E:/05.git_project/dataset/USvideos.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<value:string>

1
2
3
4
5
def csv(path: String): DataFrame = {
    // This method ensures that calls that explicit need single argument works, see SPARK-16009
    csv(Seq(path): _*)
}

1
2
type DataFrame = Dataset[Row]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
case class YoutubeVideo(video_id: String
                        , trending_date: String
                        , title: String
                        , channel_title: String
                        , category_id: String
                        , publish_time: String
                        , tags: String
                        , views: Long
                        , likes: Long
                        , dislikes: Long
                        , comment_count: String
                        , thumbnail_link: String
                        , comments_disabled: Boolean
                        , ratings_disabled: Boolean
                        , video_error_or_removed: String
                        , description: String)

case class CategoryResult(categoryId:String
                          , totalLikes:Long
                          , totalDislikes:Long)

object DataSetAnalysis {
    def main(args: Array[String]): Unit = {
        val spark = SparkSession
            .builder()
            .appName("Youtube Analysis")
            .master("local[*]")
            .config("spark.sql.shuffle.partitions",1)
            .getOrCreate()

        import spark.sqlContext.implicits._

        // 读取CSV
        val youtubeVideoDF:DataFrame = spark.read.option("header", true).csv("""E:5.git_projectdatasetUSvideos.csv""")
        // 转换为DataSet
        youtubeVideoDF.printSchema()

        // 转换为Dataset[YoutubeVideo]
        val youtubeVideoDS = youtubeVideoDF.filter(row => {
            if(row.getString(7) != null && !row.getString(7).isBlank
                && row.getString(8) != null && !row.getString(8).isBlank
                && row.getString(9) != null && !row.getString(9).isBlank) {
                if(util.Try(row.getString(7).toLong).isSuccess
                    && util.Try(row.getString(8).toLong).isSuccess
                    && util.Try(row.getString(9).toLong).isSuccess) {
                    true
                }
                else {
                    false
                }
            }
            else {
                false
            }
        })
        .map(row => YoutubeVideo(row.getString(0)
            , row.getString(1)
            , row.getString(2)
            , row.getString(3)
            , row.getString(4)
            , row.getString(5)
            , row.getString(6)
            , row.getString(7).toLong
            , row.getString(8).toLong
            , row.getString(9).toLong
            , row.getString(10)
            , row.getString(11)
            , row.getString(12).toLowerCase().toBoolean
            , row.getString(13).toLowerCase().toBoolean
            , row.getString(14)
            , row.getString(15)
        ))


        youtubeVideoDS.groupByKey(_.category_id)
            .mapValues(y => CategoryResult(y.category_id, y.likes, y.dislikes))
            .reduceGroups{(cr1, cr2) => {
                CategoryResult(cr1.categoryId, cr1.totalLikes + cr2.totalLikes, cr1.totalDislikes + cr2.totalDislikes)
            }}
            // 只获取Value部分，key部分过滤掉
            .map(t => t._2)
            .toDF()
            .show()

        TimeUnit.HOURS.sleep(1)
    }
}