SparkSQL优化
优化说明
缓存数据到内存
Spark SQL可以通过调用spark.sqlContext.cacheTable("tableName") 或者dataFrame.cache()，将表用一种柱状格式（ an inmemory columnar format）缓存至内存中。然后Spark SQL在执行查询任务时，只需扫描必需的列，从而以减少扫描数据量、提高性能。通过缓存数据，Spark SQL还可以自动调节压缩，从而达到最小化内存使用率和降低GC压力的目的。调用sqlContext.uncacheTable("tableName")可将缓存的数据移出内存。
通过sc.broadcast(spark.table("表名"))，将表广播出去，进行表与表之间的join相关操作。
可通过两种配置方式开启缓存数据功能：
1)使用spark.sqlContext的setConf方法。
2)执行SQL命令 SET key=value。
表-2 优化方式
| Property Name | Default | Meaning |
| spark.sql.inMemoryColumnarStorage.compressed | true | 如果假如设置为true，SparkSql会根据统计信息自动的为每个列选择压缩方式进行压缩 |
| spark.sql.inMemoryColumnarStorage.batchSize | 10000 | 控制列缓存的批量大小。批次大有助于改善内存使用和压缩，但是缓存数据会有OOM的风险 |
参数调优
可以通过配置下表中的参数调节Spark SQL的性能。
表-3 参数调优
| Property Name | Default | Meaning |
| spark.sql.files.maxPartitionBytes | 134217728 (128 MB) | 获取数据到分区中的最大字节数。 |
| spark.sql.files.openCostInBytes | 4194304 (4 MB) | 该参数默认4M，表示小于4M的小文件会合并到一个分区中，用于减小小文件，防止太多单个小文件占一个分区情况。 |
| spark.sql.broadcastTimeout | 300 | 广播等待超时时间，单位秒。 |
| spark.sql.autoBroadcastJoinThreshold | 10485760 (10 MB) | 最大广播表的大小。设置为-1可以禁止该功能。当前统计信息仅支持Hive Metastore表。 |
| spark.sql.shuffle.partitions | 200 | 设置shuffle分区数，默认200。 |
SQL炸裂函数
Explode：SparkSql中的列转行函数：专门针对array或map操作。
//使用explode方法必须导入下面的包：
import org.apache.spark.sql.functions._
object explode_Demo{
def main(args: Array[String]): Unit = {
//创建程序入口
val spark: SparkSession = SparkSession.builder()
.appName("createDF")
.master("local[*]")
.getOrCreate()
//调用sparkContext
val sc: SparkContext = spark.sparkContext
//设置控制台日志输出级别
sc.setLogLevel("WARN")
//导包
import spark.implicits._
//加载数据
val positionDF = spark.read.json("E:\\资料\\position.json")
//查看表结构
positionDF.printSchema()
//DSL方法处理
val listData: DataFrame = positionDF.select(explode($"data.list")).toDF("position")
//查看表结构
listData.printSchema()
//查看表数据
listData.show(false)
//查看workName并统计个数
listData.select($"position.workName" as "positions")
.groupBy($"positions")
.count()
.orderBy($"count".desc)
.show()
}
}
//SQL风格操作
/*positionDF.createOrReplaceTempView("t_position")
val sql =
"""
|select position.workName as workNames,count(*) as counts
|from(
|select explode(data.list) as position
|from t_position)
|group by workNames
|order by counts desc
""".stripMargin
spark.sql(sql).show()*/SparkSQL运行架构
Spark SQL对SQL语句的处理和关系型数据库类似，即词法/语法解析、绑定、优化、执行。Spark SQL会先将SQL语句解析成一棵树，然后使用规则(Rule)对Tree进行绑定、优化等处理过程。Spark SQL由Core、Catalyst、Hive、Hive-ThriftServer四部分构成：
1)Core: 负责处理数据的输入和输出，如获取数据，查询结果输出成DataFrame等。
2)Catalyst: 负责处理整个查询过程，包括解析、绑定、优化等。
3)Hive: 负责对Hive数据进行处理。
4)Hive-ThriftServer: 主要用于对hive的访问。
DataFrame性能上比RDD要高，主要有两方面原因：
1)定制化内存管理：Rdd数据都放在堆内存，JAVA（JVM）内存，内存管理回收分配不是由spark管理，是由JAVA（GC）管理，有时候会出现资源不一致问题，spark不是直接的内存管理者。
2)DataFrame数据以二进制的方式存在于非堆内存，节省了大量空间之外，还摆脱了GC的限制。涉及到序列化和反序列化，如图-13。
图-13 GC占比关系图
优化的执行计划
查询计划通过Spark catalyst optimiser进行优化，例子如图-14。
图-14 案例图
SparkSQL针对案例优化如图-15所示：
图-15 优化流程
为了说明查询优化，我们来看图-15展示的人口数据分析的示例。图中构造了两个DataFrame，将它们join之后又做了一次filter操作。如果原封不动地执行这个执行计划，最终的执行效率是不高的。因为join是一个代价较大的操作，也可能会产生一个较大的数据集。如果我们能将filter下推到 join下方，先对DataFrame进行过滤，再join过滤后的较小的结果集，便可以有效缩短执行时间。而Spark SQL的查询优化器正是这样做的。简而言之，逻辑查询计划优化就是一个利用基于关系代数的等价变换，将高成本的操作替换为低成本操作的过程。
得到的优化执行计划在转换成物理执行计划的过程中，还可以根据具体的数据源的特性将过滤条件下推至数据源内。最右侧的物理执行计划中Filter之所以消失不见，就是因为溶入了用于执行最终的读取操作的表扫描节点内。
对于普通开发者而言，查询优化器的意义在于，即便是经验并不丰富的程序员写出的次优的查询，也可以被尽量转换为高效的形式予以执行。
