Spark入门(六)——最全的Saprk SQL算子介绍与使用(上)

超、凢脫俗 2023-02-14 06:48 9阅读 0赞

### Spark SQL ###

*   *  Datasets & DataFrames简介
     *   *  快速入门
     *  Dataset & DataFrame实战
     *   *  Dataset create
         *   *  case-class
             *  Tuple(元组)
             *  json数据
             *  RDD
         *  Dataframe create
         *   *  json文件
             *  case-class
             *  Tuple(元组)
             *  RDD转换
         *  DataFrame Operations（Untyped）DataFrame无类型操作
         *   *  printSchema 打印Dataframe的表结构（表头）
             *  show
             *  select
             *  selectExpr
             *  withColumn
             *  withColumnRenamed
             *  drop
             *  dropDuplicates
             *  orderBy|sort
             *  groupBy
             *  agg
             *  limit
             *  where
             *  pivot（行转列）
             *  na(替换当前为null的值)
             *  join
             *  cube(多维度)
         *  Dataset Oprations (Strong typed) 数据集操作-强类型

Spark SQL是构建在RDD之上的ETL（Extract Transform Load）工具。SparkSQL在RDD之上抽象出来`Dataset/Dataframe` 这两个类提供了类似RDD的功能，也就意味用户可以使用map、faltMap、filter等高阶算子，同时也通过了基于列的命名查询，也就是说Dataset/DataFrame提供了两套操作数据的API，这些API可以给Saprk引擎要提供更多信息，系统可可以根据这些信息对计算实现一定的优化。目前Spark SQL提供了两种交互方式

1.  `SQL 脚本`
2.  `Dataset API`(strong-typed类型、untyped类型操作)

## Datasets & DataFrames简介 ##

Dataset是一个分布式数据集，Dataset是在spark-1.6提出新的API，该API构建在RDD（strong type，使用lambda表达式）之上同时可以借助于Spark SQL对执行引擎的优点，使得使用Dateset执行一些数据的转换比直接使用RDD算子功能和性能都有所提升。因此我们可以认为Dateset就是一个加强版本的RDD。 Dataset除了可以使用JVM中数组|集合对象创建之外，也可以将任意的一个RDD转换为Dataset.  
`Python does not have the support for the Dataset API.`

DataFrames 是Dataset的一种特殊情况。比如 Dataset中可以存储任意 对象类型的数据作为Dataset的元素。但是Dataframe的元素只有一种类型Row类型，这种基于Row查询和传统数据库中ResultSet操作极其相似。因为Row类型的数据表示Dataframe的一个元素，类似数据库中的一行，这些行中的元素可以通过下标或者column name访问。由于 Dateset是API的兼容或者支持度上不是多么的好，但是Dataframe在 API层面支持的Scala、Java、R、Python支持比较全面。

### 快速入门 ###

*  引入依赖

<dependency> 
    	<groupId>org.apache.spark</groupId> 
    	<artifactId>spark-core_2.11</artifactId>
        <version>2.4.3</version> 
    </dependency> 
    <dependency> 
    	<groupId>org.apache.spark</groupId> 
    	<artifactId>spark-sql_2.11</artifactId> 
    	<version>2.4.3</version> 
    </dependency>

*  创建字符统计（untyped）

def main(args: Array[String]): Unit = { 
    		//1.创建SparkSeesion
    		val spark = SparkSession
    			.builder()
    			.appName("wordcount")
    			.master("local[6]")
    			.getOrCreate()
    
    		//2.导入spark定义隐式增强|转换
    		import spark.implicits._
    
    		//3.创建dataset
    		val lines = Array("this is a demo", "hello spark")
    		val wordRDD = spark.sparkContext
    			.makeRDD(lines)
    			.flatMap(_.split("\\s+"))
    			.map((_, 1))
    
    		val ds: Dataset[(String, Int)] = wordRDD.toDS()
    		
    		//4.对Dataset执行sql算子操作
    		ds.groupBy($"_1")
    			//无类型操作
    			.sum("_2")
    			.as("total")
    			.withColumnRenamed("_1", "word")
    			.withColumnRenamed("sum(_2)", "total")
    			.show()
    		//5.关闭spark
    		spark.stop()
    }

*  创建字符统计（strong typed）

def main(args: Array[String]): Unit = { 
    		//1.创建SparkSeesion 
    		val spark = SparkSession
    			.builder()
    			.appName("wordcount")
    			.master("local[6]")
    			.getOrCreate()
    
    		//2.导入spark定义隐式增强|转换
    
    		import spark.implicits._
    
    		//3.创建dataset 
    		val lines = Array("this is a demo", "hello spark")
    		val wordRDD = spark.sparkContext
    			.makeRDD(lines)
    			.flatMap(_.split("\\s+"))
    			.map((_, 1))
    		val ds: Dataset[(String, Int)] = wordRDD.toDS()
    
    		//4.对Dataset执行sql算子操作 
    		ds.groupByKey(t => t._1)
    			.agg(typed.sum[(String, Int)](tuple => tuple._2).name("total"))
    			.show()
    
    		//5.关闭spark 
    		spark.stop()
    }

## Dataset & DataFrame实战 ##

### Dataset create ###

Dataset类似于RDD，不同的是Spark SQL有一套自己的序列化规范独立于Spark RDD（Java/Kryo序列化）之上称为Encoders。不同于SparkRDD序列化，由于Dataset支持无类型操作，用户无需获取操作的类型，操作仅仅是列名，因为Spark SQL在执行算子操作的时候可以省略反序列化的步骤，继而提升程序执行  
效率。

#### case-class ####

// case-class
    case class Person(id: Int, name: String, age: Int, sex: Boolean)

/** * case-class */
    val person: Dataset[Person] = List(
       Person(1, "zhangsan", 18, true),
       Person(2, "lisi", 28, true)
    )
       .toDS()
    person.select($"id", $"name")
       .show()

> 注：
> 
>  *  加入隐式转换 import spark.implicits.\_ 该语句需要放在获取spark对象的语句之后
>  *  case class Person(id: Int, name: String, age: Int, sex: Boolean) 的定义需要放在方法的作用域之外（即`Scala/Java`的成员变量位置）
>  *  如果case类是java编写，则java bean需要实现 `Serializable`，以及增加`get、set`方法

结果:

+---+--------+
    | id|    name|
    +---+--------+
    |  1|zhangsan|
    |  2|    lisi|
    +---+--------+

#### Tuple(元组) ####

/** * tuple */
    val person: Dataset[(Int, String, Int, Boolean)] = List(
    	(1, "zhangsan", 18, true),
    	(2, "lisi", 28, true)
    )
    	.toDS()
    person.select($"_1", $"_2").show()

结果:

+---+--------+
    | _1|      _2|
    +---+--------+
    |  1|zhangsan|
    |  2|    lisi|
    +---+--------+

#### json数据 ####

数据：

{"name":"张三","age":18}
    {"name":"李四","age":28}
    {"name":"王五","age":38}

//数值默认是long类型
    case class User(name: String, age: Long)

/** * json格式 */
    spark.read
    	.json("file:///Users/mashikang/IdeaProjects/spark_sql/src/main/resources/json/")
    	.as[User]
    	.show()

结果:

+---+----+
    |age|name|
    +---+----+
    | 18|张三|
    | 28|李四|
    | 38|王五|
    +---+----+

#### RDD ####

*  Tuple(元组)

/** * RDD 元组 */
    val userRDD = spark.sparkContext.makeRDD(List((1,"张三",true,18,15000.0)))
    userRDD.toDS().show()

结果:

+---+----+----+---+-------+
    | _1|  _2|  _3| _4|     _5|
    +---+----+----+---+-------+
    |  1|张三|true| 18|15000.0|
    +---+----+----+---+-------+

*  case-class

// case-class
    case class Person(id: Int, name: String, age: Int, sex: Boolean)

/** * RDD case-class */
    val userRDD = spark.sparkContext.makeRDD(List(Person(1,"张三",18,true)))
    userRDD.toDS().show()

结果:

+---+----+---+----+
    | id|name|age| sex|
    +---+----+---+----+
    |  1|张三| 18|true|
    +---+----+---+----+

### Dataframe create ###

DataFrame是一个命名列的数据集，用户可以直接操作 `column` 因此几乎所有DataFrame推荐操作都是`无类型操作` 。用户也可以把一个DataFrame看做是 `Dataset[Row]` 类型的数据集。

#### json文件 ####

/** * json */
    val dataFrame: DataFrame = spark.read
    	.json("file:///Users/mashikang/IdeaProjects/spark_sql/src/main/resources/json/")
    dataFrame.printSchema()
    dataFrame.show()

结果:

root
     |-- age: long (nullable = true)
     |-- name: string (nullable = true)
    
    +---+----+
    |age|name|
    +---+----+
    | 18|张三|
    | 28|李四|
    | 38|王五|
    +---+----+

#### case-class ####

case class User(id:Int,name:String,sex:Boolean)

/** * case-class */
    var userDF=List(User(1,"张三",true))
    	.toDF()
    userDF.show()

结果: