Spark之SparkSQL实战

布满荆棘的人生 2022-01-23 03:03 403阅读 0赞

DataFrames 基本操作和 DSL SQL风格 UDF函数 以及数据源：

## SparkSQL查询 ##

Json数据准备

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 </pre></td> 
 <td><pre>{
 "name":"Michael"} {
 "name":"Andy", "age":30} {
 "name":"Justin", "age":19} </pre></td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 </pre></td> 
 <td><pre>val df =spark.read.json("/input/sparksql/json/people.json") df.show() df.filter($"age"&gt;21).show(); df.createOrReplaceTempView("person") spark.sql("SELECT * FROM person").show() </pre></td> 
 </tr> 
 </tbody> 
</table>

![20190530221923.png][]

## IDEA创建SparkSQL程序 ##

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 </pre></td> 
 <td><pre>&lt;dependency&gt; &lt;groupId&gt;org.apache.spark&lt;/groupId&gt; &lt;artifactId&gt;spark-sql_2.11&lt;/artifactId&gt; &lt;version&gt;2.1.1&lt;/version&gt; &lt;scope&gt;provided&lt;/scope&gt; &lt;/dependency&gt; </pre></td> 
 </tr> 
 </tbody> 
</table>

![20190531121301.png][]

## SparkSession ##

老的版本中，SparkSQL提供两种SQL查询起始点，一个叫SQLContext，用于Spark自己提供的SQL查询，一个叫HiveContext，用于连接Hive的查询，SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext，所以计算实际上是由sparkContext完成的。

![20190601104448.png][]

## 创建DataFrames ##

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

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 </pre></td> 
 <td><pre>val df = spark.read.json("/input/sparksql/people.json")	 df.show() val peopleRdd = sc.textFile("/input/sparksql/people.txt") val peopleDF = peopleRdd.map(_.split(",")).map(paras =&gt; (paras(0),paras(1).trim().toInt)).toDF("name","age") peopleDF.show() </pre></td> 
 </tr> 
 </tbody> 
</table>

![20190601123919.png][]

### 常用操作 ###

### DSL风格语法 ###

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 </pre></td> 
 <td><pre>df.printSchema() df.select("name").show() df.select($"name", $"age" + 1).show() df.filter($"age" &gt; 21).show() df.groupBy("age").count().show() </pre></td> 
 </tr> 
 </tbody> 
</table>

![20190601124441.png][]

### SQL风格语法 ###

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 </pre></td> 
 <td><pre>df.createOrReplaceTempView("people") val sqlDF = spark.sql("SELECT * FROM people") sqlDF.show() df.createGlobalTempView("people") spark.sql("SELECT * FROM global_temp.people").show() spark.newSession().sql("SELECT * FROM global_temp.people").show() </pre></td> 
 </tr> 
 </tbody> 
</table>

临时表是Session范围内的，Session退出后，表就失效了。如果想应用范围内有效，可以使用全局表。注意使用全局表时需要全路径访问，如：global\_temp.people

![20190601124700.png][]

## 创建DataSet ##

注意: Case classes in Scala 2.10只支持 22 字段. 你可以使用自定义的 classes 来实现对字段的映射  
`case class Person(name: String, age: Long)`

![20190601151413.png][]

## 相互转化 ##

具体的转换可以参考: [三者共性][Link 1]

## UDF函数 ##

![20190601162725.png][]

## 自定义聚合函数 ##

强类型的Dataset和弱类型的DataFrame都提供了相关的聚合函数， 如 count()，countDistinct()，avg()，max()，min()。除此之外，用户可以设定自己的自定义聚合函数。

弱类型用户自定义聚合函数：通过继承`UserDefinedAggregateFunction`来实现用户自定义聚

<table> 
 <tbody> 
 <tr> 
 <td><pre>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 </pre></td> 
 <td><pre>package com.hph.sql import org.apache.spark.sql.{
 Row, SparkSession} import org.apache.spark.sql.expressions.{
 MutableAggregationBuffer, UserDefinedAggregateFunction} import org.apache.spark.sql.types._ //自定义UDAF函数需要继承UserDefinedAggregateFunction class AverageSal extends UserDefinedAggregateFunction {
 //输入数据 override def inputSchema: StructType = StructType(StructField("salary", LongType) :: Nil) //定义每一个分区中的共享变量 override def bufferSchema: StructType = StructType(StructField("sum", LongType) :: StructField("count", IntegerType) :: Nil) //表示UDAF函数的输出类型 override def dataType: DataType = DoubleType //如果有相同的输入是否会存在相同的输出,如果是则为True override def deterministic: Boolean = true //初始化 每一个分区中的共享变量 override def initialize(buffer: MutableAggregationBuffer): Unit = {
 buffer(0) = 0L; // sum buffer(1) = 0 // sum } //每一个分区中的每一条数据聚合的时候需要调用该方法 override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
 //获取这一行中的工资,然后将工资加入到sum里 buffer(0) = buffer.getLong(0) + input.getLong(0) //需要将工资的个数加1 buffer(1) = buffer.getInt(1) + 1 } //将每一个没去的输出合并形成最后的数据 override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
 //合并总的工资 buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0) //合并总的工资个数 buffer1(1) = buffer1.getInt(1) + buffer2.getInt(1) } //给出计算结果 override def evaluate(buffer: Row): Any = {
 //取出总的工资 / 总的工资个数 buffer.getLong(0).toDouble / buffer.getInt(1) } } object AverageSal {
 def main(args: Array[String]): Unit = {
 //创建SparkConf()并设置App名称 val spark = SparkSession.builder() .appName("Spark SQL UDF example") .master("local[*]") .getOrCreate() val employee = spark.read.json("hdfs://datanode1:9000/input/sparksql/employees.json") employee.createOrReplaceTempView("employee") spark.udf.register("average", new AverageSal) spark.sql("select average(salary) from employee").show() spark.stop() } } </pre></td> 
 </tr> 
 </tbody> 
</table>

### 强类型函数 ###

<table> 
 <tbody> 
 <tr> 
 <td><pre>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 </pre></td> 
 <td><pre>package com.hph.sql import org.apache.spark.sql.{
 Encoder, Encoders, SparkSession} import org.apache.spark.sql.expressions.Aggregator case class Employee(name: String, salary: Long) case class Aver(var sum: Long, var count: Int) class Average extends Aggregator[Employee, Aver, Double] {
 //初始化方法:初始化每一个分区的共享变量 override def zero: Aver = Aver(0L, 0) //每一个分区的每一条数据聚合的时候需要回调该方法 override def reduce(b: Aver, a: Employee): Aver = {
 b.sum = b.sum + a.salary b.count = b.count + 1 b } //将每一个分区的输出 合并 形成最后的数据 override def merge(b1: Aver, b2: Aver): Aver = {
 b1.sum = b1.sum + b2.sum b1.count = b1.count + b2.count b1 } //输出计算结果 override def finish(reduction: Aver): Double = {
 reduction.sum.toDouble / reduction.count } override def bufferEncoder: Encoder[Aver] = Encoders.product override def outputEncoder: Encoder[Double] = Encoders.scalaDouble } object Average {
 def main(args: Array[String]): Unit = {
 //创建SparkConf()并设置App名称 val spark = SparkSession.builder() .appName("Spark SQL Strong Type UDF example") .master("local[*]") .getOrCreate() import spark.implicits._ val employee = spark.read.json("hdfs://datanode1:9000/input/sparksql/employees.json").as[Employee] val aver = new Average().toColumn.name("average") employee.select(aver).show() spark.close() } } </pre></td> 
 </tr> 
 </tbody> 
</table>

![20190601200128.png][]

[20190530221923.png]: /images/20220123/f624052fc7a04170b8c9f24538cde302.png
[20190531121301.png]: /images/20220123/1dc0ef9a160b4ddf8477a2c876da5d47.png
[20190601104448.png]: /images/20220123/92425e37bc024d8ba29f150149cd241b.png
[20190601123919.png]: /images/20220123/0b739278841d4f8c824bf24130ea80b6.png
[20190601124441.png]: /images/20220123/3cd3906897c74130ba74fe938dc52609.png
[20190601124700.png]: /images/20220123/b66b119e22c1457085218c0810dc47ac.png
[20190601151413.png]: /images/20220123/a608fd2d7a5d4fa2bc18e6a0afb3398b.png
[Link 1]: https://www.hphblog.cn/2019/05/30/Spark%E4%B9%8BSparkSQL/#%E4%B8%89%E8%80%85%E5%85%B1%E6%80%A7
[20190601162725.png]: /images/20220123/6530195c7d22412486a693e5a029838c.png
[20190601200128.png]: /images/20220123/bc856d0a029d48d6be64334914a1f1c9.png