① Spark sql一列拆分多列
将DataFrame中的一列拆分为多列，示例如下：
② Spark SQL CBO 基于代价的优化
Spark CBO 背景
本文将介绍 CBO，它充分考虑了数据本身的特点（如大小、分布）以及操作算子的特点（中间结果集的分布及大小）及代价，从而更好的选择执行代价最小的物理执行计划，即 SparkPlan。
Spark CBO 原理
CBO 原理是计算所有可能的物理计划的代价，并挑选出代价最小的物理执行计划。其核心在于评估一个给定的物理执行计划的代价。
物理执行计划是一个树状结构，其代价等于每个执行节点的代价总合，如下图所示。
而每个执行节点的代价，分为两个部分
每个操作算子的代价相对固定，可用规则来描述。而执行节点输出数据集的大小与分布，分为两个部分：1) 初始数据集，也即原始表，其数据集的大小与分布可直接通过统计得到；2)中间节点输出数据集的大小与分布可由其输入数据集的信息与操作本身的特点推算。
所以，最终主要需要解决两个问题
Statistics 收集
通过如下 SQL 语句，可计算出整个表的记录总数以及总大小
从如下示例中，Statistics 一行可见， customer 表数据总大小为 37026233 字节，即 35.3MB，总记录数为 28万，与事实相符。
通过如下 SQL 语句，可计算出指定列的统计信息
从如下示例可见，customer 表的 c_customer_sk 列最小值为 1， 最大值为 280000，null 值个数为 0，不同值个数为 274368，平均列长度为 8，最大列长度为 8。
除上述示例中的统计信息外，Spark CBO 还直接等高直方图。在上例中，histogram 为 NULL。其原因是，spark.sql.statistics.histogram.enabled 默认值为 false，也即 ANALYZE 时默认不计算及存储 histogram。
下例中，通过 SET spark.sql.statistics.histogram.enabled=true; 启用 histogram 后，完整的统计信息如下。
从上图可见，生成的 histogram 为 equal-height histogram，且高度为 1102.36，bin 数为 254。其中 bin 个数可由 spark.sql.statistics.histogram.numBins 配置。对于每个 bin，匀记录其最小值，最大值，以及 distinct count。
值得注意的是，这里的 distinct count 并不是精确值，而是通过 HyperLogLog 计算出来的近似值。使用 HyperLogLog 的原因有二
算子对数据集影响估计
对于中间算子，可以根据输入数据集的统计信息以及算子的特性，可以估算出输出数据集的统计结果。
本节以 Filter 为例说明算子对数据集的影响。
对于常见的 Column A < value B Filter，可通过如下方式估算输出中间结果的统计信息
上述估算的前提是，字段 A 数据均匀分布。但很多时候，数据分布并不均匀，且当数据倾斜严重是，上述估算误差较大。此时，可充分利用 histogram 进行更精确的估算
启用 Historgram 后，Filter Column A < value B 的估算方法为
在上图中，B.value = 15，A.min = 0，A.max = 32，bin 个数为 10。Filter 后 A.ndv = ndv(<B.value) = ndv(<15)。该值可根据 A < 15 的 5 个 bin 的 ndv 通过 HyperLogLog 合并而得，无须重新计算所有 A < 15 的数据。
算子代价估计
SQL 中常见的操作有 Selection（由 select 语句表示），Filter（由 where 语句表示）以及笛卡尔乘积（由 join 语句表示）。其中代价最高的是 join。
Spark SQL 的 CBO 通过如下方法估算 join 的代价
其中 rows 即记录行数代表了 CPU 代价，size 代表了 IO 代价。weight 由 *spark.sql.cbo.joinReorder.card.weight *决定，其默认值为 0.7。
Build侧选择
对于两表Hash Join，一般选择小表作为build size，构建哈希表，另一边作为 probe side。未开启 CBO 时，根据表原始数据大小选择 t2 作为build side
而开启 CBO 后，基于估计的代价选择 t1 作为 build side。更适合本例
优化 Join 类型
Spark SQL 中，Join 可分为 Shuffle based Join 和 BroadcastJoin。Shuffle based Join 需要引入 Shuffle，代价相对较高。BroadcastJoin 无须 Join，但要求至少有一张表足够小，能通过 Spark 的 Broadcast 机制广播到每个 Executor 中。
在不开启 CBO 中，Spark SQL 通过 spark.sql.autoBroadcastJoinThreshold 判断是否启用 BroadcastJoin。其默认值为 10485760 即 10 MB。
并且该判断基于参与 Join 的表的原始大小。
在下图示例中，Table 1 大小为 1 TB，Table 2 大小为 20 GB，因此在对二者进行 join 时，由于二者都远大于自动 BroatcastJoin 的阈值，因此 Spark SQL 在未开启 CBO 时选用 SortMergeJoin 对二者进行 Join。
而开启 CBO 后，由于 Table 1 经过 Filter 1 后结果集大小为 500 GB，Table 2 经过 Filter 2 后结果集大小为 10 MB 低于自动 BroatcastJoin 阈值，因此 Spark SQL 选用 BroadcastJoin。
优化多表 Join 顺序
未开启 CBO 时，Spark SQL 按 SQL 中 join 顺序进行 Join。极端情况下，整个 Join 可能是 left-deep tree。在下图所示 TPC-DS Q25 中，多路 Join 存在如下问题，因此耗时 241 秒。
开启 CBO 后， Spark SQL 将执行计划优化如下
优化后的 Join 有如下优势，因此执行时间降至 71 秒
总结
5万人关注的大数据成神之路，不来了解一下吗？
5万人关注的大数据成神之路，真的不来了解一下吗？
5万人关注的大数据成神之路，确定真的不来了解一下吗？
③ Spark Sql 函数使用
round - 保留数据精度
如 round(col("col1"),0) 对应数值为 21.23 -> 21.0 ；21.73 -> 22.0
如 round(col("col1"),1) 对应数值为 21.23 -> 21.2
如 round(col("col1"),-1) 对应数值为 21.23 -> 20.0
④ 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
⑤ Learning Spark [6] - Spark SQL高级函数
collect常用的局磨有两个函数：collect_list(不去重)和collect_set(去重)
collect_list
collect_set
explode的定义是将桐缓斗数组的每个数据展开，如下我们就可以将上面的dataframe还原为最初的样式。哪罩
posexplode可以在拆分列的同时，增加一列序号
但是如果表内有如下两个一一对应的数组，我们该如何拆分呢？
按照直觉，我们尝试分别explode()
解决这个问题，我们需要使用 LATERAL VIEW
lateral view可以理解为创建了一个表，然后JOIN到了查询的表上，这样就避免了两个生成器的问题
split则是将一个字符串根据分隔符，变化为一个数组
transform会引用一个函数在数组的每个元素上，返回一个数列
filter为通过条件删选，返回一个数列
exists为判断是否包含该元素，返回一个布尔值
rece为通过两个函数，将数组聚合为一个值，然后对该值进行运算
Reference
Learning Spark 2nd - Lightning Fast Big Data Analysis by Jules S. Damji, Brooke Wenig, Tathagata Das, and Denny Lee
⑥ spark sql启动后执行越来越慢是为什么
Shark为了实现Hive兼容，在HQL方面森闷重用了Hive中HQL的解析、逻辑执行计划翻译、执行计划优化等逻辑，可以近似认为仅将物理执行计划从MR作业替换成了Spark作业（辅以内存列式存储等各种和Hive关系不大的优化）；同时还依赖Hive Metastore和Hive SerDe（用于兼容现有的各种Hive存储格式）。这一策略导致了两个问题，第一是执行计划优化完全依赖于Hive，不方便昌简添加新的优化策略；二是因为MR是进程级并行，写代码的时候不是很注意线程安全问题，导致Shark不得不使用另外一套独立维耐春裤护的打了补丁的Hive源码分支（至于为何相关修改没有合并到Hive主线，我也不太清楚）。
此外，除了兼容HQL、加速现有Hive数据的查询分析以外，Spark SQL还支持直接对原生RDD对象进行关系查询。同时，除了HQL以外，Spark SQL还内建了一个精简的SQL parser，以及一套Scala DSL。也就是说，如果只是使用Spark SQL内建的SQL方言或Scala DSL对原生RDD对象进行关系查询，用户在开发Spark应用时完全不需要依赖Hive的任何东西。
⑦ 大数据如何入门
首先我们要了解Java语言和Linux操作系统，这两个是学习大数据的基础，学习的顺序不分前后。
大数据
Java ：只要了解一些基础即可，做大数据不需要很深的Java 技术，学java SE 就相当于有学习大数据基础。
Linux：因为大数据相关软件都是在Linux上运行的，所以Linux要学习的扎实一些，学好Linux对你快速掌握大数据相关技术会有很大的帮助，能让你更好的理解hadoop、hive、hbase、spark等大数据软件的运行环境和网络环境配置，能少踩很多坑，学会shell就能看懂脚本这样能更容易理解和配置大数据集群。还能让你对以后新出的大数据技术学习起来更快。
Hadoop：这是现在流行的大数据处理平台几乎已经成为大数据的代名词，所以这个是必学的。Hadoop里面包括几个组件HDFS、MapRece和YARN，HDFS是存储数据的地方就像我们电脑的硬盘一样文件都存储在这个上面，MapRece是对数据进行处理计算的，它有个特点就是不管多大的数据只要给它时间它就能把数据跑完，但是时间可能不是很快所以它叫数据的批处理。
Zookeeper：这是个万金油，安装Hadoop的HA的时候就会用到它，以后的Hbase也会用到它。它一般用来存放一些相互协作的信息，这些信息比较小一般不会超过1M，都是使用它的软件对它有依赖，对于我们个人来讲只需要把它安装正确，让它正常的run起来就可以了。
Mysql：我们学习完大数据的处理了，接下来学习学习小数据的处理工具mysql数据库，因为一会装hive的时候要用到，mysql需要掌握到什么层度那?你能在Linux上把它安装好，运行起来，会配置简单的权限，修改root的密码，创建数据库。这里主要的是学习SQL的语法，因为hive的语法和这个非常相似。
Sqoop：这个是用于把Mysql里的数据导入到Hadoop里的。当然你也可以不用这个，直接把Mysql数据表导出成文件再放到HDFS上也是一样的，当然生产环境中使用要注意Mysql的压力。
Hive：这个东西对于会SQL语法的来说就是神器，它能让你处理大数据变的很简单，不会再费劲的编写MapRece程序。有的人说Pig那?它和Pig差不多掌握一个就可以了。
Oozie：既然学会Hive了，我相信你一定需要这个东西，它可以帮你管理你的Hive或者MapRece、Spark脚本，还能检查你的程序是否执行正确，出错了给你发报警并能帮你重试程序，最重要的是还能帮你配置任务的依赖关系。我相信你一定会喜欢上它的，不然你看着那一大堆脚本，和密密麻麻的crond是不是有种想屎的感觉。
Hbase：这是Hadoop生态体系中的NOSQL数据库，他的数据是按照key和value的形式存储的并且key是唯一的，所以它能用来做数据的排重，它与MYSQL相比能存储的数据量大很多。所以他常被用于大数据处理完成之后的存储目的地。
Kafka：这是个比较好用的队列工具，队列是干吗的?排队买票你知道不?数据多了同样也需要排队处理，这样与你协作的其它同学不会叫起来，你干吗给我这么多的数据(比如好几百G的文件)我怎么处理得过来，你别怪他因为他不是搞大数据的，你可以跟他讲我把数据放在队列里你使用的时候一个个拿，这样他就不在抱怨了马上灰流流的去优化他的程序去了，因为处理不过来就是他的事情。而不是你给的问题。当然我们也可以利用这个工具来做线上实时数据的入库或入HDFS，这时你可以与一个叫Flume的工具配合使用，它是专门用来提供对数据进行简单处理，并写到各种数据接受方(比如Kafka)的。
Spark：它是用来弥补基于MapRece处理数据速度上的缺点，它的特点是把数据装载到内存中计算而不是去读慢的要死进化还特别慢的硬盘。特别适合做迭代运算，所以算法流们特别稀饭它。它是用scala编写的。Java语言或者Scala都可以操作它，因为它们都是用JVM的。
⑧ 如何避免Spark SQL做数据导入时产生大量小文件
生产上，我们往往将Spark SQL作为Hive的替代方案，来获得SQL on Hadoop更出色的性能。因此，本文所讲的是指存储于HDFS中小文件，即指文件的大小远小于HDFS上块（dfs.block.size）大小的文件。
比如我们拿TPCDS测试集中的store_sales进行举例， sql如下所示
首先我们得到其执行计划，如下所示，
store_sales的原生文件包含1616逻辑分片，对应生成1616 个Spark Task，插入动态分区表之后生成1824个数据分区加一个NULL值的分区，每个分区下都有可能生成1616个文件，这种情况下，最终的文件数量极有可能达到2949200。1T的测试集store_sales也就大概300g，这种情况每个文件可能就零点几M。
比如，为了防止Shuffle阶段的数据倾斜我们可以在上面的sql中加上 distribute by rand() ，这样我们的执行计划就变成了，
这种情况下，这样我们的文件数妥妥的就是spark.sql.shuffle.partitions * N，因为rand函数一般会把数据打散的非常均匀。当spark.sql.shuffle.partitions设置过大时，小文件问题就产生了；当spark.sql.shuffle.partitions设置过小时，任务的并行度就下降了，性能随之受到影响。
最理想的情况，当然是根据分区字段进行shuffle，在上面的sql中加上 distribute by ss_sold_date_sk 。 把同一分区的记录都哈希到同一个分区中去，由一个Spark的Task进行写枣租察入，这样的话只会产生N个文件，在我们的case中store_sales，在1825个分区下各型简种生成了一个数据文件。
但是这种情况下也容易出现数据倾斜的问题，比如双11的销售数据就很容易在这种情况下发生倾斜。
前面已经提到根据分区字段进行分区，除非每个分区下本身的数据较少，分区字段选择不合理，那么小文件问题基本上就不存在了，但是也有可能由于shuffle引入新的数据倾斜问题。
我们首先可以尝试是否可以将两者结合使用， 在之前的sql上加上 distribute by ss_sold_date_sk，cast(rand() * 5 as int) ， 这个类似于我们处理数据倾斜问题时候给字段加上后缀的形式。如，
按照之前的推算，每个分区下将产生5个文件，同时null值倾斜部分的数据也被打散成五份进行计算，缓解了数据倾斜的问题 ，我们最终将得到1825 *5=9105个文件，如下所示
如果我们将5改得更小，文件数也会越少，但相应的倾斜key的计算时间也会上去。
在我们知道那个分区键倾斜的情况下，我们也可以将入库的SQL拆成几个部分，比如我们store_sales是因为null值倾斜，我们就可以通过 where ss_sold_date_sk is not null 和 where ss_sold_date_sk is null 将原始数据分成两个部分。前者可以基于分区字段进行分区，如 distribute by ss_sold_date_sk ;后者可以基于随机值进行分区， distribute by cast(rand() * 5 as int) , 这样可以静态的将null值部分分成五个文件。
对于倾斜部分的数据，我们可以开启Spark SQL的自适应功能， spark.sql.adaptive.enabled=true 来动态调整每个相当于Spark的rece端task处理的数据量，这样我们就不需要人为的感知随机值的规模了，我们可以直接
然后Spark在Shuffle 阶段会自动的帮我们将数据尽量的凳茄合并成 spark.sql.adaptive.shuffle.targetPostShuffleInputSize （默认64m）的大小，以减少输出端写文件线程的总量，最后减少个数。
对于 spark.sql.adaptive.shuffle.targetPostShuffleInputSize 参数而言，我们也可以设置成为 dfs.block.size 的大小，这样可以做到和块对齐，文件大小可以设置的最为合理。
在我们的 猛犸大数据平台 上面，随便的建立几个SQL作业，不用会Spark也可以用SQL把大数据玩得666！
双击每个工作节点，我们也可以对我们的SQL作业进行参数的调整
选中我们对应的实验组，点击执行后，可以查看任务的运行状态。
从各组的实验结果来看
实验组一的小文件控制还是可喜可贺的。对于我们1t的tpcds测试数据，null值分区字段下只有40个文件，其他每个数据分区也只有一个数据文件，总目录1825，总文件数1863. 在解决数据倾斜问题的基础上，也只比纯按照分区字段进行distibute by多了39个文件。
本文讲述的是如何在纯写SQL的场景下，如何用Spark SQL做数据导入时候，控制小文件的数量。
⑨ 如何使用 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的风险