Ⅰ 如何使用 Spark sql
一、启动方法
/data/spark-1.4.0-bin-cdh4/bin/spark-sql --master spark://master:7077 --total-executor-cores 10 --executor-memory 1g --executor-cores 2
注：/data/spark-1.4.0-bin-cdh4/为spark的安装路径
/data/spark-1.4.0-bin-cdh4/bin/spark-sql –help 查看启动选项
--master MASTER_URL 指定master url
--executor-memory MEM 每个executor的内存，默认为1G
--total-executor-cores NUM 所有executor的总核数
-e <quoted-query-string> 直接执行查询SQL
-f <filename> 以文件方式批量执行SQL
二、Spark sql对hive支持的功能
1、查询语句：SELECT GROUP BY ORDER BY CLUSTER BY SORT BY
2、hive操作运算：
1) 关系运算：= ==, <>, <, >, >=, <=
2) 算术运算：+, -, *, /, %
3) 逻辑运算：AND, &&, OR, ||
4) 复杂的数据结构
5) 数学函数：(sign, ln, cos, etc)
6) 字符串函数：
3、 UDF
4、 UDAF
5、 用户定义的序列化格式
6、join操作：JOIN {LEFT|RIGHT|FULL} OUTER JOIN LEFT SEMI JOIN CROSS JOIN
7、 unions操作：
8、 子查询： SELECT col FROM ( SELECT a + b AS col from t1) t2
9、Sampling
10、 Explain
11、 分区表
12、 视图
13、 hive ddl功能：CREATE TABLE、CREATE TABLE AS SELECT、ALTER TABLE
14、 支持的数据类型：TINYINT SMALLINT INT BIGINT BOOLEAN FLOAT DOUBLE STRING BINARY TIMESTAMPDATE ARRAY MAP STRUCT
三、Spark sql 在客户端编程方式进行查询数据
1、启动spark-shell
./spark-shell --master spark://master:7077 --total-executor-cores 10 --executor-memory 1g --executor-cores 2
2、编写程序
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val df = sqlContext.read.json("../examples/src/main/resources/people.json")
查看所有数据：df.show()
查看表结构：df.printSchema()
只看name列：df.select("name").show()
对数据运算：df.select(df("name"), df("age") + 1).show()
过滤数据：df.filter(df("age") > 21).show()
分组统计：df.groupBy("age").count().show()
1、查询txt数据
import sqlContext.implicits._
case class Person(name: String, age: Int)
val people = sc.textFile("../examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
people.registerTempTable("people")
val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")
2、parquet文件
val df = sqlContext.read.load("../examples/src/main/resources/users.parquet")
3、hdfs文件
val df = sqlContext.read.load("hdfs://namenode.Hadoop:9000/user/hive/warehouse/spark_test.db/test_parquet/part-r-00001.gz.parquet")
4、保存查询结果数据
val df = sqlContext.read.load("../examples/src/main/resources/users.parquet")
df.select("name", "favorite_color").write.save("namesAndFavColors.parquet“)
四、Spark sql性能调优
缓存数据表：sqlContext.cacheTable("tableName")
取消缓存表：sqlContext.uncacheTable("tableName")
spark.sql.inMemoryColumnarStorage.compressedtrue当设置为true时，Spark SQL将为基于数据统计信息的每列自动选择一个压缩算法。
spark.sql.inMemoryColumnarStorage.batchSize10000柱状缓存的批数据大小。更大的批数据可以提高内存的利用率以及压缩效率，但有OOMs的风险
Ⅱ Spark-sql读取hive分区表限制分区过滤条件及限制分区数量
在开发过程中使用spark去读取hive分区表的过程中（或者使用hive on spark、nodepad开发工具），部分开发人员未注意添加分区属性过滤导致在执行过程中加载了全量数据，引起任务执行效率低、磁盘IO大量损耗等问题
1、自定义规则CheckPartitionTable类，实现Rule
然后通过此种方法创建SparkSession
2、自定义规则CheckPartitionTable类，实现Rule，将规则类追加致Optimizer.batches: Seq[Batch]中
1、CheckPartitionTable规则执行类，需要通过引入sparkSession从而获取到引入conf；需要继承Rule[LogicalPlan]；
2、通过splitPredicates方法，分离分区谓词，得到分区谓词表达式
在sql解析过程中将谓词解析为TreeNode，此处采用递归的方式获取分区谓词
3、判断是否是分区表，且是否添加分区字段
4、实现Rule的apply方法
关于spark-sql的主要执行流程及预备知识，可参照我同学的这篇博文 https://www.jianshu.com/p/4cc6797fb9ce
Ⅲ spark sql dataset怎么做分组排序呢
sparksql怎样取分组后的topn
Spark SQL 开窗函数
1、Spark 1.5.x版本以后，在Spark SQL和DataFrame中引入了开窗函数，比如最经典的就是我们的row_number()，可以让我们实现分组取topn的逻辑。
2、做一个案例进行topn的取值（利用Spark的开窗函数），不知道是否还有印象，我们之前在最早的时候，做过topn的计算，当时是非常麻烦的。但是现在用了Spark SQL之后，非常方便。
Ⅳ Ku：Spark SQL操作Ku
摘要： Spark SQL ， Ku
参考 https://github.com/xieenze/SparkOnKu/blob/master/src/main/scala/com/spark/test/KuCRUD.scala
引入 spark-core_2.11 ， spark-sql_2.11 ， ku-spark2_2.11 ， hadoop-client 依赖包
指定 ku.master" ， ku.table ，如果读取超时加入 ku.operation.timeout.ms 参数
或者
写入数据可以使用dataframe的 write 方法，也可以使用 kuContext 的 updateRows ， insertRows ， upsertRows ， insertIgnoreRows 方法
直接调用dataframe的write方法指定 ku.master ， ku.table ，只支持 append 模式，对已有key的数据自动更新
调用kuContext的 upsertRows 方法，效果和dataframe调用write append模式一样
调用kuContext insertRows ， insertIgnoreRows 方法，如果插入的数据key已存在insertRows直接报错，insertIgnoreRows忽略已存在的key，只插入不存在的key
调用kuContext updateRows 方法，对已经存在的key数据做更新，如果key不存在直接报错
使用已有dataframe的schema建表
使用 StructType 自定义schema
删除表和判断表是否存在
Ⅳ Spark SQL（十）：Hive On Spark
Hive是目前大数据领域，事实上的SQL标准。其底层默认是基于MapRece实现的，但是由于MapRece速度实在比较慢，因此这几年，陆续出来了新的SQL查询引擎，包括Spark SQL，Hive On Tez，Hive On Spark等。
Spark SQL与Hive On Spark是不一样的。Spark SQL是Spark自己研发出来的针对各种数据源，包括Hive、JSON、Parquet、JDBC、RDD等都可以执行查询的，一套基于Spark计算引擎的查询引擎。因此它是Spark的一个项目，只不过提供了针对Hive执行查询的工功能而已，适合在一些使用Spark技术栈的大数据应用类系统中使用。
而Hive On Spark，是Hive的一个项目，它是将Spark作为底层的查询引擎（不通过MapRece作为唯一的查询引擎）。Hive On Spark，只适用于Hive，在可预见的未来，很有可能Hive默认的底层引擎就从MapRece切换为Spark了；适合于将原有的Hive数据仓库以及数据统计分析替换为Spark引擎，作为全公司通用的大数据统计分析引擎。
Hive On Spark做了一些优化：
1、Map Join
Spark SQL默认对join是支持使用broadcast机制将小表广播到各个节点上，以进行join的。但是问题是，这会给Driver和Worker带来很大的内存开销。因为广播的数据要一直保留在Driver内存中。所以目前采取的是，类似乎MapRece的Distributed Cache机制，即提高HDFS replica factor的复制因子，以让数据在每个计算节点上都有一个备份，从而可以在本地进行数据读取。
2、Cache Table
对于某些需要对一张表执行多次操作的场景，Hive On Spark内部做了优化，即将要多次操作的表cache到内存中，以便于提升性能。但是这里要注意，并不是对所有的情况都会自动进行cache。所以说，Hive On Spark还有很多不完善的地方。
Hive QL语句 =>
语法分析 => AST =>
生成逻辑执行计划 => Operator Tree =>
优化逻辑执行计划 => Optimized Operator Tree =>
生成物理执行计划 => Task Tree =>
优化物理执行计划 => Optimized Task Tree =>
执行优化后的Optimized Task Tree
Ⅵ spark SQL和hive到底什么关系
Hive是一种基于HDFS的数据仓库，并且提供了基于SQL模型的，针对存储了大数据的数据仓库，进行分布式交互查询的查询引擎。
SparkSQL并不能完全替代Hive，它替代的是Hive的查询引擎，SparkSQL由于其底层基于Spark自身的基于内存的特点，因此速度是Hive查询引擎的数倍以上，Spark本身是不提供存储的，所以不可能替代Hive作为数据仓库的这个功能。
SparkSQL相较于Hive的另外一个优点，是支持大量不同的数据源，包括hive、json、parquet、jdbc等等。SparkSQL由于身处Spark技术堆栈内，基于RDD来工作，因此可以与Spark的其他组件无缝整合使用，配合起来实现许多复杂的功能。比如SparkSQL支持可以直接针对hdfs文件执行sql语句。
Ⅶ 可能是全网最详细的 Spark Sql Aggregate 源码剖析
纵观 Spark Sql 源码，聚合的实现是其中较为复杂的部分，本文希望能以例子结合流程图的方式来说清楚整个过程。这里仅关注 Aggregate 在物理执行计划相关的内容，之前的 parse、analyze 及 optimize 阶段暂不做分析。在 Spark Sql 中，有一个专门的 Aggregation strategy 用来处理聚合，我们先来看看这个策略。
本文暂不讨论 distinct Aggregate 的实现（有兴趣的可以看看另一篇博文 https://www.jianshu.com/p/77e0a70db8cd ），我们来看看 AggUtils#planAggregateWithoutDistinct 是如何生成聚合的物理执行计划的
创建聚合分为两个阶段：
AggregateExpression 共有以下几种 mode：
Q：是否支持使用 hash based agg 是如何判断的？
摘自我另一篇文章： https://www.jianshu.com/p/77e0a70db8cd
为了说明最常用也是最复杂的的 hash based agg，本小节暂时将示例 sql 改为
这样就能进入 HashAggregateExec 的分支
构造函数主要工作就是对 groupingExpressions、aggregateExpressions、aggregateAttributes、resultExpressions 进行了初始化
在 enable code gen 的情况下，会调用 HashAggregateExec#inputRDDs 来生成 RDD，为了分析 HashAggregateExec 是如何生成 RDD 的，我们设置 spark.sql.codegen.wholeStage 为 false 来 disable code gen，这样就会调用 HashAggregateExec#doExecute 来生成 RDD，如下：
可以看到，关键的部分就是根据 child.execute() 生成的 RDD 的每一个 partition 的迭代器转化生成一个新的 TungstenAggregationIterator ，即 HashAggregateExec 生成的 RDD 的各个 partition。由于 TungstenAggregationIterator 涉及内容非常多，我们单开一大节来进行介绍。
此迭代器：
注：UnsafeKVExternalSorter 的实现可以参考：
UnsafeRow 是 InternalRow（表示一行记录） 的 unsafe 实现，由原始内存（byte array）而不是 Java 对象支持，由三个区域组成：
使用 UnsafeRow 的收益：
构造函数的主要流程已在上图中说明，需要注意的是：当内存不足时（毕竟每个 grouping 对应的 agg buffer 直接占用内存，如果 grouping 非常多，或者 agg buffer 较大，容易出现内存用尽）会从 hash based aggregate 切换为 sort based aggregate（会 spill 数据到磁盘），后文会进行详述。先来看看最关键的 processInputs 方法的实现
上图中，需要注意的是：hashMap 中 get 一个 groupingKey 对应的 agg buffer 时，若已经存在该 buffer 则直接返回；若不存在，尝试申请内存新建一个：
上图中，用于真正处理一条 row 的 AggregationIterator#processRow 还需进一步展开分析。在此之前，我们先来看看 AggregateFunction 的分类
AggregateFunction 可以分为 DeclarativeAggregate 和 ImperativeAggregate 两大类，具体的聚合函数均为这两类的子类。
DeclarativeAggregate 是一类直接由 Catalyst 中的 Expressions 构成的聚合函数，主要逻辑通过调用 4 个表达式完成，分别是：
我们再次以容易理解的 Count 来举例说明：
通常来讲，实现一个基于 Expressions 的 DeclarativeAggregate 函数包含以下几个重要的组成部分：
再来看看 AggregationIterator#processRow
AggregationIterator#processRow 会调用
生成用于处理一行数据（row）的函数
说白了 processRow 生成了函数才是直接用来接受一条 input row 来更新对应的 agg buffer，具体是根据 mode 及 aggExpression 中的 aggFunction 的类型调用其 updateExpressions 或 mergeExpressions 方法：
比如，对于 aggFunction 为 DeclarativeAggregate 类型的 Partial 下的 Count 来说就是调用其 updateExpressions 方法，即：
对于 Final 的 Count 来说就是调用其 mergeExpressions 方法，即：
对于 aggFunction 为 ImperativeAggregate 类型的 Partial 下的 Collect 来说就是调用其 update 方法，即：
对于 Final 的 Collect 来说就是调用其 merge 方法，即：
我们都知道，读取一个迭代器的数据，是要不断调用 hasNext 方法进行 check 是否还有数据，当该方法返回 true 的时候再调用 next 方法取得下一条数据。所以要知道如何读取 TungstenAggregationIterator 的数据，就得分析其这两个方法。
分为两种情况，分别是：
Agg 的实现确实复杂，本文虽然篇幅已经很长，但还有很多方面没有 cover 到，但基本最核心、最复杂的点都详细介绍了，如果对于未 cover 的部分有兴趣，请自行阅读源码进行分析~