1. Spark SQL概述
1.1 什么是Spark SQL
Spark SQL是用于结构化数据处理的Spark模块。与基本的Spark RDD API不同，Spark SQL提供的接口为Spark提供了有关数据结构和正在执行的计算的更多信息。在内部，Spark SQL使用这些额外的信息来执行额外的优化。与Spark SQL交互的方式有多种，包括SQL和Dataset API。计算结果时，使用相同的执行引擎，与您用于表达计算的API/语言无关。
1.2 为什么要有Spark SQL
1.3 SparkSQL的发展
1）发展历史
RDD（Spark1.0）=》Dataframe（Spark1.3）=》Dataset（Spark1.6）
如果同样的数据都给到这三个数据结构，他们分别计算之后，都会给出相同的结果。不同的是他们的执行效率和执行方式。在现在的版本中，dataSet性能最好，已经成为了唯一使用的接口。其中Dataframe已经在底层被看做是特殊泛型的DataSet<Row>。
2）三者的共性
（1）RDD、DataFrame、DataSet全都是Spark平台下的分布式弹性数据集，为处理超大型数据提供便利。
（2）三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action行动算子如foreach时，三者才会开始遍历运算。
（3）三者有许多共同的函数，如filter，排序等。
（4）三者都会根据Spark的内存情况自动缓存运算。
（5）三者都有分区的概念。
1.4 Spark SQL的特点
1）易整合
无缝的整合了SQL查询和Spark编程。
2）统一的数据访问方式
使用相同的方式连接不同的数据源。
3）兼容Hive
在已有的仓库上直接运行SQL或者HQL。
4）标准的数据连接
通过JDBC或者ODBC来连接。
1.5 SparkSession新的起始点
在老的版本中，SparkSQL提供两种SQL查询起始点：
- 一个叫SQLContext，用于Spark自己提供的SQL查询；
- 一个叫HiveContext，用于连接Hive的查询。
SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。
SparkSession内部封装了SparkContext，所以计算实际上是由SparkContext完成的。当我们使用spark-shell的时候，Spark框架会自动的创建一个名称叫做Spark的SparkSession，就像我们以前可以自动获取到一个sc来表示SparkContext。
[atguigu@hadoop102 spark-local]$ bin/spark-shell
20/09/12 11:16:35 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://hadoop102:4040
Spark context available as 'sc' (master = local[*], app id = local-1599880621394).
Spark session available as 'spark'.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 3.3.1
/_/
Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_212)
Type in expressions to have them evaluated.
Type :help for more information.
2 常用方式
2.1 方法调用
1）创建一个maven工程SparkSQL
2）创建包名为com.atguigu.sparksql
3）输入文件夹准备：在新建的SparkSQL项目名称上右键=》新建input文件夹=》在input文件夹上右键=》新建user.json。并输入如下内容：
{"age":20,"name":"qiaofeng"}{"age":19,"name":"xuzhu"}{"age":18,"name":"duanyu"}
{"age":22,"name":"qiaofeng"}
{"age":11,"name":"xuzhu"}
{"age":12,"name":"duanyu"}
5）在pom.xml文件中添加spark-sql的依赖
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.3.1</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.22</version>
</dependency>
</dependencies>
6）代码实现
添加javaBean的User
package com.atguigu.sparksql.Bean;
import lombok.Data;
import java.io.Serializable;
@Data
public class User implements Serializable {
public Long age;
public String name;
public User() {
}
public User(Long age, String name) {
this.age = age;
this.name = name;
}
}
代码编写
package com.atguigu.sparksql;
import com.atguigu.sparksql.Bean.User;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.api.java.function.ReduceFunction;
import org.apache.spark.sql.*;
import scala.Tuple2;
public class Test01_Method {
public static void main(String[] args) {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
// 按照行读取
Dataset<Row> lineDS = spark.read().json("input/user.json");
// 转换为类和对象
Dataset<User> userDS = lineDS.as(Encoders.bean(User.class));
// userDS.show();
// 使用方法操作
// 函数式的方法
Dataset<User> userDataset = lineDS.map(new MapFunction<Row, User>() {
@Override
public User call(Row value) throws Exception {
return new User(value.getLong(0), value.getString(1));
}
},
// 使用kryo在底层会有部分算子无法使用
Encoders.bean(User.class));
// 常规方法
Dataset<User> sortDS = userDataset.sort(new Column("age"));
sortDS.show();
// 区分
RelationalGroupedDataset groupByDS = userDataset.groupBy("name");
// 后续方法不同
Dataset<Row> count = groupByDS.count();
// 推荐使用函数式的方法 使用更灵活
KeyValueGroupedDataset<String, User> groupedDataset = userDataset.groupByKey(new MapFunction<User, String>() {
@Override
public String call(User value) throws Exception {
return value.name;
}
}, Encoders.STRING());
// 聚合算子都是从groupByKey开始
// 推荐使用reduceGroup
Dataset<Tuple2<String, User>> result = groupedDataset.reduceGroups(new ReduceFunction<User>() {
@Override
public User call(User v1, User v2) throws Exception {
// 取用户的大年龄
return new User(Math.max(v1.age, v2.age), v1.name);
}
});
result.show();
//4. 关闭sparkSession
spark.close();
}
}
在sparkSql中DS直接支持的转换算子有：map（底层已经优化为mapPartition）、mapPartition、flatMap、groupByKey（聚合算子全部由groupByKey开始）、filter、distinct、coalesce、repartition、sort和orderBy（不是函数式的算子，不过不影响使用）。
2.2 SQL使用方式
package com.atguigu.sparksql;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
public class Test02_SQL {
public static void main(String[] args) {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
Dataset<Row> lineDS = spark.read().json("input/user.json");
// 创建视图 => 转换为表格 填写表名
// 临时视图的生命周期和当前的sparkSession绑定
// orReplace表示覆盖之前相同名称的视图
lineDS.createOrReplaceTempView("t1");
// 支持所有的hive sql语法,并且会使用spark的又花钱
Dataset<Row> result = spark.sql("select * from t1 where age > 18");
result.show();
//4. 关闭sparkSession
spark.close();
}
}}
2.3 DSL特殊语法（扩展）
package com.atguigu.sparksql;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.col;
public class Test03_DSL {
public static void main(String[] args) {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
// 导入特殊的依赖 import static org.apache.spark.sql.functions.col;
Dataset<Row> lineRDD = spark.read().json("input/user.json");
Dataset<Row> result = lineRDD.select(col("name").as("newName"),col("age").plus(1).as("newAge"))
.filter(col("age").gt(18));
result.show();
//4. 关闭sparkSession
spark.close();
}
}
3 SQL语法的用户自定义函数
3.1 UDF 用户自定义函数
1）UDF：一行进入，一行出
2）代码实现
package com.atguigu.sparksql;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.expressions.UserDefinedFunction;
import org.apache.spark.sql.types.DataTypes;
import static org.apache.spark.sql.functions.udf;
public class Test04_UDF {
public static void main(String[] args) {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
Dataset<Row> lineRDD = spark.read().json("input/user.json");
lineRDD.createOrReplaceTempView("user");
// 定义一个函数
// 需要首先导入依赖import static org.apache.spark.sql.functions.udf;
UserDefinedFunction addName = udf(new UDF1<String, String>() {
@Override
public String call(String s) throws Exception {
return s + " 大侠";
}
}, DataTypes.StringType);
spark.udf().register("addName",addName);
spark.sql("select addName(name) newName from user")
.show();
// lambda表达式写法
spark.udf().register("addName1",(UDF1<String,String>) name -> name + " 大侠",DataTypes.StringType);
//4. 关闭sparkSession
spark.close();
}
}
3.2 UDAF 用户自定义聚合函数
1）UDAF：输入多行，返回一行。通常和groupBy一起使用，如果直接使用UDAF函数，默认将所有的数据合并在一起。
2）Spark3.x推荐使用extends Aggregator自定义UDAF，属于强类型的Dataset方式。
3）Spark2.x使用extends UserDefinedAggregateFunction，属于弱类型的DataFrame
4）案例实操
需求：实现求平均年龄，自定义UDAF，MyAvg(age)
（1）自定义聚合函数实现-强类型
package com.atguigu.sparksql;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.expressions.Aggregator;
import java.io.Serializable;
import static org.apache.spark.sql.functions.udaf;
public class Test05_UDAF {
public static void main(String[] args) {
//1. 创建配置对象
SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");
//2. 获取sparkSession
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
//3. 编写代码
spark.read().json("input/user.json").createOrReplaceTempView("user");
// 注册需要导入依赖 import static org.apache.spark.sql.functions.udaf;
spark.udf().register("avgAge",udaf(new MyAvg(),Encoders.LONG()));
spark.sql("select avgAge(age) newAge from user").show();
//4. 关闭sparkSession
spark.close();
}
public static class Buffer implements Serializable {
private Long sum;
private Long count;
public Buffer() {
}
public Buffer(Long sum, Long count) {
this.sum = sum;
this.count = count;
}
public Long getSum() {
return sum;
}
public void setSum(Long sum) {
this.sum = sum;
}
public Long getCount() {
return count;
}
public void setCount(Long count) {
this.count = count;
}
}
public static class MyAvg extends Aggregator<Long,Buffer,Double>{
@Override
public Buffer zero() {
return new Buffer(0L,0L);
}
@Override
public Buffer reduce(Buffer b, Long a) {
b.setSum(b.getSum() + a);
b.setCount(b.getCount() + 1);
return b;
}
@Override
public Buffer merge(Buffer b1, Buffer b2) {
b1.setSum(b1.getSum() + b2.getSum());
b1.setCount(b1.getCount() + b2.getCount());
return b1;
}
@Override
public Double finish(Buffer reduction) {
return reduction.getSum().doubleValue() / reduction.getCount();
}
@Override
public Encoder<Buffer> bufferEncoder() {
// 可以用kryo进行优化
return Encoders.kryo(Buffer.class);
}
@Override
public Encoder<Double> outputEncoder() {
return Encoders.DOUBLE();
}
}
}
3.3 UDTF（没有）
输入一行，返回多行（Hive）。
SparkSQL中没有UDTF，需要使用算子类型的flatMap先完成拆分。
