Spark之Spark Session、Dataframe、Dataset

亦凉 2022-11-27 08:49 205阅读 0赞

## Spark SQL简介 ##

**Spark SQL架构：**  
Spark SQL是Spark的核心组件之一（2014.4 Spark1.0)

*  能够直接访问现存的Hive数据
 *  提供JDBC/ODBC接口供第三方工具借助Spark进行数据处理
 *  提供了更高层级的接口方便地处理数据
 *  支持多种操作方式：SQL、API编程
 *  支持多种外部数据源：Parquet、JSON、RDBMS等

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]

Spark SQL是Spark的一个模块，主要用于处理结构化的数据。与基础的Spark RDD API不同的是，Spark SQL的接口会向提供更多的信息，包括数据结构以及要执行的计算操作等。在Spark SQL内部，会使用这些信息执行一些额外的优化。使用Spark SQL有两种方式，包括SQL语句以及Dataset API。  
但是在计算的时候，无论你是用哪种接口去进行计算，它们使用的底层执行引擎是完全一模一样的。这种底层执行机制的统一，就意味着我们可以在不同的方式之间任意来回切换，只要我们可以灵活地运用不同的方式来最自然地表达我们要执行的计算操作就可以了。

## 对其中SQL的解释 ##

Spark SQL的一个主要的功能就是执行SQL查询语句。Spark 2.0开始，最大的一个改变，就是支持了SQL 2003标准语法，还有就是支持子查询。

Spark SQL也可以用来从Hive中查询数据。当我们使用某种编程语言开发的Spark作业来执行SQL时，返回的结果是Dataframe/Dataset类型的。当然，我们也可以通过Spark SQL的shell命令行工具，或者是JDBC/ODBC接口来访问。

## Spark SQL之DataSet简介 ##

Dataset是一个分布式的数据集，指特定域对象中的强类型集合。Dataset是Spark 1.6开始新引入的一个接口，它结合了RDD API的很多优点（包括强类型，支持lambda表达式等），以及Spark SQL的优点（优化后的执行引擎）。  
Dataset可以通过JVM对象来构造，然后通过transformation类算子（map，flatMap，filter等）来进行操作。  
Scala和Java的API中支持Dataset，但是Python不支持Dataset API。不过因为Python语言本身的天然动态特性，Dataset API的不少feature本身就已经具备了（比如可以通过row.columnName来直接获取某一行的某个字段）。

#### DataSet三种基本创建方式 ####

scala> spark.createDataset(1 to 3)
    
    scala> spark.createDataset(List(("a",1),("b",2),("c",3)))
    
    scala> spark.createDataset(sc.parallelize(List(("a",1,1),("b",2,2))))

**注意：**

1.  createDataset()的参数可以是：Seq、Array、RDD
2.  上面三行代码生成的Dataset分别是：  
    Dataset\[Int\]、Dataset\[(String,Int)\]、Dataset\[(String,Int,Int)\]
3.  Dataset=RDD+Schema，所以Dataset与RDD有大部共同的函数，如map、filter等

**使用Case Class创建Dataset**

case class Point(label:String,x:Double,y:Double)
    val points=Seq(Point("bar",3.0,5.6),Point("foo",1.0,3.0)).toDS

## Spark SQL之DataFrame简介 ##

Dataframe就是按列组织的Dataset。在逻辑概念上，可以大概认为Dataframe等同于关系型数据库中的表，或者是Python/R语言中的data frame，但是在底层做了大量的优化。  
Dataframe可以通过很多方式来构造：比如结构化的数据文件，Hive表，数据库，已有的RDD。  
Scala，Java，Python，R等语言都支持Dataframe。  
在Scala API中，Dataframe就是Dataset\[Row\]的类型别名。在Java中，需要使用Dataset来代表一个Dataframe。

**DataFrame=Dataset\[Row\]**  
类似传统数据的二维表格。  
在RDD基础上加入了Schema（数据结构信息）  
DataFrame Schema支持嵌套数据类型

*  struct
 *  map
 *  array

#### DataFrame创建方式 ####

**使用createDataFrame创建**

spark.createDataFrame(List( ("Scala", 35),("Python", 30), ("R", 15), ("Java", 20)) )

**通过spark.read读取json文件，生成dataframe**

spark.read.json("file:///data/language.json")

## SparkSession ##

从Spark 2.0开始，一个最大的改变就是，Spark SQL的统一入口就是SparkSession，SQLContext和HiveContext未来会被淘汰。可以通过SparkSession.builder()来创建一个SparkSession。  
SparkSession内置就支持Hive，包括使用HiveQL语句查询Hive中的数据，使用Hive的UDF函数，以及从Hive表中读取数据等。

val spark = SparkSession
    	.builder()
    	.appName("SparkTest")
    	.master("local[*]") 
    	.getOrCreate()
    
    import spark.implicits._

## SparkSession：运行SQL查询 ##

SparkSession的sql()函数允许我们执行SQL语句，得到的结果是一个Dataframe。

val df = spark.read.json("file:///data/language.json")
    df.createOrReplaceTempView("language")
    val sqlDF = spark.sql("SELECT * FROM language")
    sqlDF.show()

## Dataframe的untyped操作 ##

有了SparkSession之后，就可以通过已有的RDD，Hive表，或者其他数据源来创建Dataframe.  
比如说通过json文件来创建。Dataframe提供了一种domain-specific language来进行结构化数据的操作，这种操作也被称之为untyped操作，与之相反的是基于强类型的typed操作。

val df = spark.read.json("file:///data/language.json")
    df.printSchema()
    df.show()
    df.select("name").show()
    df.select($"name", $"hours" + 1).show()
    df.filter($"hours" > 21).show()
    df.groupBy("hours").count().show()

## Dataset的typed操作 ##

Dataset与RDD比较类似，但是非常重要的一点不同是，RDD的序列化机制是基于Java序列化机制或者是Kryo的，而Dataset的序列化机制基于一种特殊的Encoder，来将对象进行高效序列化，以进行高性能处理或者是通过网络进行传输。

Dataset除了Encoder，也同时支持Java序列化机制，但是encoder的特点在于动态的代码生成，同时提供一种特殊的数据格式，来让spark不将对象进行反序列化，即可直接基于二进制数据执行一些常见的操作，比如filter、sort、hash等。

case class Point(label:String,x:Double,y:Double)
    val points=Seq(Point("bar",3.0,5.6),Point("foo",1.0,3.0)).toDS
    points.show()
    
    val primitiveDS = Seq(1, 2, 3).toDS()
    primitiveDS.map(_ + 1).collect()

## Spark SQL写parquet文件 ##

Parquet文件：sparksql默认数据源，是一种流行的列式存储格式，以二进制存储，文件中包含数据与元数据

import org.apache.spark.sql.types.{ StructType, StructField, StringType,ArrayType,IntegerType}
    val schema=StructType(Array(StructField("name",StringType),
    			    StructField("favorite_color",StringType),
                 		    StructField("favorite_numbers",ArrayType(IntegerType))))
    val rdd=sc.parallelize(List(("Alyssa",null,Array(3,9,15,20)),("Ben","red",null)))
    val rowRDD=rdd.map(p=>Row(p._1,p._2,p._3))
    val df=spark.createDataFrame(rowRDD,schema)
    
    #在该目录下生成parquet文件
    df.write.parquet("/data/users")

**注意：**  
StructType中的StructField字段个数和类型要与parquet文件中的一致

## Spark SQL读parquet文件 ##

//该目录下存在parquet文件  
val df=spark.read.parquet("/data/users")  
df.show  
df.printSchema  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1][]  
Spark Session、Dataframe、Dataset的创建以及使用就暂时说到这里，后面会详细介绍三者优劣及区别，以及如何相互转换。记得关注小编！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/1aa02f510fc046f48ef198cf10d2f282.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/ae8490def15c473bbc00090387be83df.png