Spark-SQL——DataFrame与Dataset_java datasets spark.sql

一、Spark SQL概述

1.1、Spark SQL是什么？

Spark SQL是Apache Spark中用于结构化数据处理的模块。它允许开发人员在Spark上执行SQL查询、处理结构化数据以及将它们与常规的RDD一起使用。Spark Sql提供了用于处理结构化数据的高级API，如DataFrames和Datasets，它们比原始的RDD API更加高效和方便。
通过Spark SQL，可以使用标准的SQL语言进行数据处理，也可以使用DataFrame API进行操作。此外，Spark SQL还可以集成Hive，允许用户直接查询Hive表并将查询结果存储到Hive中。Spark SQL也支持Spark中运行标准的JDBC/ODBC连接，从而可以使用各种外部数据源。

1.2、Hive和Spark SQL

Hive和Spark SQL都是用来处理大数据的工具，主要是基于Hadoop生态圈。它们的相同点都是用来查询和处理大规模数据的，而且都可以使用类SQL语言来进行操作。

不同之处：

• 操作语言不同：Hive使用HQL(Hive Query Language)进行数据操作，而Spark SQL使用Spark SQL语法进行数据操作。
• 数据处理方式不同：Hive依赖于MapReduce作为计算引擎，而Spark SQL使用Spark作为计算引擎。由于Spark的内存计算能力较强，所以在某些场景下，Spark SQL比Hive更加高效。
• 数据存储方式不同：Hive使用HDFS或者其他支持Hadoop HDFS API的存储系统来存储数据，而Spark SQL可以支持多种不同的数据存储系统，例如：HDFS、Hive等。
• 性能不同：Spark SQL的性能要比Hive快得多，主要是因为Spark Sql使用了内存计算技术，而Hive使用的是MapReduce计算模型。
• 执行计划不同：Hive的执行计划是通过HQL生成的，而Spark Sql的执行计划是通过Spark的优化器生成的。Spark的优化器可以对查询进行优化，以提高查询的性能。

1.3、DataFrame与DataSet

• DataFrame
  • DataFrame是Spark SQL的一种数据抽象，它表示分布式数据集合。DataFrame和关系型数据库中的表类似，都有列和行的概念，而且还具备了分布式的特性。DataFrame提供了丰富的数据操作接口，例如：选择、过滤、分组、聚合、排序、连接等。这些操作和SQL语句类似，便于使用和理解。
  • DataFrame可以看作是由一系列Row对象组成的RDD，每个Row代表着一行数据。不同于RDD，DataFrame具有数据类型的概念，这意味着DataFrame的每一列都有其对应的数据类型，这一特性使得Spark能够进行更为智能的优化，例如通过列式存储来提升查询效率。同时DataFrame的API使用了强类型语言的特性，使得在编译期就能够捕获到许多错误，提高了开发效率。
• Dataset
  • Dataset是Spark中的一种强类型API，它结合了Spark SQL中的DataFrame和RDD的优点。
  • Dataset API提供了编译时类型安全和面向对象的编程接口，并利用Spark SQL的优化执行引擎来提高查询性能。它可以使用Scala、Java和Python等语言进行操作，支持多种数据源，如Parquet、JSON、JDBC等。
  • Dataset API还提供了对关系型数据和面向对象数据的支持，可将其转换为Dataset，通过统一的编程接口进行处理和操作。
• 二者不同：
  • 数据类型：DataFrame 是一种以列为基础的分布式数据集合，强调数据的结构，即 schema，可以使用 SQL 或者 DataFrame 的 API 进行操作；Dataset 是一种类型化的分布式数据集合，强调类型的安全性，可以使用强类型的 API 进行操作。
  • 类型检查：DataFrame 只有在运行时才会进行类型检查，可能在运行时出现类型错误；Dataset 利用 Spark 强大的编译器来捕捉类型错误，可以在编译时就发现类型错误。
  • API：DataFrame 的 API 是基于 DataFrame 和 SQL 的 API，使用较为灵活，但是对于类型的检查较为弱；Dataset 的 API 是基于 Scala 和 Java 的编程语言，可以保证类型的安全性，但是使用的时候会相对复杂一些。
  • 编译时优化：Dataset 可以受益于 Spark 的 Catalyst 查询优化器，在编译时进行优化，可以提高性能；DataFrame 只有在运行时才能够获得优化。
  • 语言支持：DataFrame 可以使用 Scala，Java，Python 和 R 等编程语言，比 Dataset 支持的编程语言更多。Dataset 只支持 Scala 和 Java。
  • 性能：由于 Dataset 强制类型检查，所以可以获得更好的性能，因此在需要高性能的场景下推荐使用 Dataset。但是对于某些特定场景下，DataFrame 可能会更适合。

二、Spark SQL编程

Spark Core 中，如果想要执行应用程序，需要首先构建上下文环境对象 SparkContext，Spark SQL 其实可以理解为对 Spark Core 的一种封装，不仅仅在模型上进行了封装，上下文环境对象也进行了封装。
SparkSession 是 Spark 最新的 SQL 查询起始点，实质上是 SQLContext 和 HiveContext的组合，所以在 SQLContext 和 HiveContext 上可用的 API 在 SparkSession 上同样是可以使用的。SparkSession 内部封装了 SparkContext，所以计算实际上是由 sparkContext 完成的。当我们使用 spark-shell 的时候, spark 框架会自动的创建一个名称叫做 spark 的 SparkSession 对象, 就像我们以前可以自动获取到一个 sc 来表示 SparkContext 对象一样

2.1、DataFrame

2.1.1、创建DataFrame

在Spark SQL中SparkSession是创建DataFrame和执行SQL的入口，创建DataFrame有三种方式：通过Spark的数据源进行创建；从一个存在的RDD进行转换；还可以从HiveTable进行查询返回。

• 从Spark数据源进行创建
  • Spark支持的创建文件的数据源格式：
    
  • 创建文件user.json文件
    {“username”:“zhangsan”,“age”:20}
  • 读取json文件创建DataFrame

    scala> spark.read.json("file:///opt/stufile/user.json")
    res2: org.apache.spark.sql.DataFrame = [age: bigint, username: string]          
    1
    2

  • 结果

    scala> res2.show()
    +---+--------+
    |age|username|
    +---+--------+
    | 20|zhangsan|
    +---+--------+
    1
    2
    3
    4
    5
    6

• RDD与DataFrame互转
  • 在IDEA中开发程序时，如果需要RDD与DF或者DS之间进行互相操作，那么需要引入 import spark.implicits._
  • 在spark-shell中无需导入，自动完成此操作
  • 创建样例类

    scala> case class User(name:String,age:Int)
    defined class User
    1
    2

  • 创建RDD

    sc.makeRDD(List(("zhangsan",30),("lisi",20)))
    res4: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[8] at makeRDD at <console>:25
    
    1
    2
    3

  • 转换DataFrame

    scala> res5.map(t => User(t._1,t._2)).toDF
    res7: org.apache.spark.sql.DataFrame = [name: string, age: int]
    
    scala> res7.show
    +--------+---+
    |    name|age|
    +--------+---+
    |zhangsan| 30|
    |    lisi| 20|
    +--------+---+
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11

  • 转换为RDD

    scala> res7.rdd
    res9: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[21] at rdd at <console>:26
    
    1
    2
    3

2.1.2、SQL语法

SQL语法风格是指我们查询数据的时候使用SQL语句来查询，这种风格的查询必须要有临时视图或者全局视图来辅助。

df.createOrReplaceTempView(“people”) 创建临时表（创建或重写临时表）
df.createOrReplaceGlobalTempView(“people”) 创建全局表（创建或重写全局表）
**注意：**普通临时表是Session范围内的，如果想要应用范围内有效，可以使用全局临时表。使用全局临时表是需要全路径访问，前面加上global_temp. 例如：select * from global_temp.people

• 读取JSON文件创建DataFrame

scala> val df = spark.read.json("file:///opt/stufile/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, username: string] 
1
2

• 对DataFrame创建一个临时表

scala> df.createOrReplaceTempView("people")
1

• 通过SQL语句实现查询全表

scala> val sqlDF = spark.sql("SELECT * FROM people")
sqlDF: org.apache.spark.sql.DataFrame = [age: bigint， name: string]
1
2

• 结果展示

scala> sqlDF.show
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+
1
2
3
4
5
6

• 创建全局表

scala> df.createOrReplaceGlobalTempView("people")
1

• 查询全表

scala> spark.sql("SELECT * FROM global_temp.people").show()
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+
scala> spark.newSession().sql("SELECT * FROM global_temp.people").show()
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+
1
2
3
4
5
6
7
8
9
10
11
12

2.1.3、DSL语法

DataFrame提供一个特定领域语言（domain-specific language,DSL）去管理结构化的数据。可以在Scala、Java、Python等语言中使用DSL，使用DSL语法风格不必去创建临时表。

• 创建一个DataFrame

  scala> val df = spark.read.json("file:///opt/stufile/user.json")
  df: org.apache.spark.sql.DataFrame = [age: bigint, username: string] 
     
  1
  2
  3

• 查看DataFrame的Schema信息

  	scala> df.printSchema
  	root
  	 |-- age: long (nullable = true)
  	 |-- username: string (nullable = true)
  1
  2
  3
  4

• 只查看username列数据

  scala> df.select("username").show()
  +--------+
  |username|
  +--------+
  |zhangsan|
  | lisi|
  | wangwu|
  +--------+
  1
  2
  3
  4
  5
  6
  7
  8

• 查看username列数据以及age+1数据

  scala> df.select($"username",$"age" + 1).show
  scala> df.select('username, 'age + 1).show()
  scala> df.select('username, 'age + 1 as "newage").show()
  +--------+---------+
  |username|(age + 1)|
  +--------+---------+
  |zhangsan| 21|
  | lisi| 31|
  | wangwu| 41|
  +--------+---------+
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

• 查看age大于30的数据

  scala> df.filter($"age">30).show
  +---+---------+
  |age| username|
  +---+---------+
  | 40| wangwu|
  +---+---------+
  1
  2
  3
  4
  5
  6

• 按照age分组，查看数据条数

  scala> df.groupBy("age").count.show
  +---+-----+
  |age|count|
  +---+-----+
  | 20| 1|
  | 30| 1|
  | 40| 1|
  +---+-----+
  1
  2
  3
  4
  5
  6
  7
  8

2.2、Dataset

DataSet是具有强类型的数据集合，需要提供对于的类型信息。

2.2.1、创建DataSet

• 使用样例类序列创建Dataset

  scala> case class Person(name: String, age: Long)
  defined class Person
  scala> val caseClassDS = Seq(Person("zhangsan",2)).toDS()
  caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: Long]
  scala> caseClassDS.show
  +---------+---+
  | name|age|
  +---------+---+
  | zhangsan| 2|
  +---------+---+
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

• 使用基本类型的序列创建Dataset

  scala> val ds = Seq(1,2,3,4,5).toDS
  ds: org.apache.spark.sql.Dataset[Int] = [value: int]
  scala> ds.show
  +-----+
  |value|
  +-----+
  | 1|
  | 2|
  | 3|
  | 4|
  | 5|
  +-----+
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12

2.2.2、RDD与Dataset互转

• RDD转换Dataset

  SparkSQL 能够自动将包含有 case 类的 RDD 转换成 DataSet，case 类定义了 table 的结构，case 类属性通过反射变成了表的列名。Case 类可以包含诸如 Seq 或者 Array 等复杂的结构。

  scala> case class User(name:String, age:Int)
  defined class User
  
  scala> sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1, t._2)).toDS
  res0: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
  
  scala> res0.show
  +--------+---+                                                                  
  |    name|age|
  +--------+---+
  |zhangsan| 30|
  |    lisi| 49|
  +--------+---+
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14

• Dataset转换为RDD

  scala> case class User(name:String, age:Int)
  defined class User
  scala> sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1,t._2)).toDS
  res11: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
  scala> val rdd = res11.rdd
  rdd: org.apache.spark.rdd.RDD[User] = MapPartitionsRDD[51] at rdd at <console>:25
  scala> rdd.collect
  res12: Array[User] = Array(User(zhangsan,30), User(lisi,49))
  1
  2
  3
  4
  5
  6
  7
  8

2.2.3、DataFrame 和 DataSet 转换

• DataFrame 转换为 DataSet

  scala> case class User(name:String, age:Int)
  defined class User
  scala> val df = sc.makeRDD(List(("zhangsan",30),
  ("lisi",49))).toDF("name","age")
  df: org.apache.spark.sql.DataFrame = [name: string, age: int]
  scala> val ds = df.as[User]
  ds: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
  1
  2
  3
  4
  5
  6
  7

• DataSet 转换为 DataFrame

  scala> val ds = df.as[User]
  ds: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
  scala> val df = ds.toDF
  df: org.apache.spark.sql.DataFrame = [name: string, age: int]
  1
  2
  3
  4