sparkSQL1.1入门之六：sparkSQL之基础应用 - 数据库综合 - 次元立方网 - 电脑知识与技术互动交流平台
sparkSQL1.1入门之六：sparkSQL之基础应用
sparkSQL1.1对数据的查询分成了2个分支：sqlContext 和 hiveContext。
在sqlContext中，sparkSQL可以使用SQL-92语法对定义的表进行查询，表的源数据可以来自：
RDDparquet文件json文件
在hiveContext中，sparkSQL可以使用HQL语法，对hive数据进行查询，sparkSQL1.1支持hive0.12的HQL语法；如果遇上不支持的语法，用户可以通过更改配置切换到sql语法。笔者猜测，从spark1.1开始，将打开sqlContext和hiveContext之间的壁垒，混用sqlContext和hiveContext中定义的表。本文因时间关系，尚未对该特性做测试，等spark1.1正式版发布，再专门讨论一下这个问题。另外，在hiveContext中，hql()将被弃用，sql()将代替hql()来提交查询语句。
为了方便演示，我们在spark-shell里面进行下列演示，并加以说明。首先，启动spark集群，然后在客户端wy上启动spark-shell：
bin/spark-shell --master spark://hadoop1:7077 --executor-memory 3g
1：sqlContext基础应用
sqlContext先将外部读入的数据转换成SchemaRDD，然后注册成表，才能进行表的操作。要使用sqlContext，首先要引入sqlContext库及其隐式函数：
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext._
要将RDD转换成sqlContext中的table，首先要定义case class，在RDD的transform过程中使用case class可以隐式转化成SchemaRDD，然后再使用registerAsTable注册成表。注册成表后就可以在sqlContext对表进行操作，如select 、insert、join等。注意，case class可以是嵌套的，也可以使用类似Sequences 或 Arrays之类复杂的数据类型。
下面的例子是定义一个符合数据文件/sparksql/people.txt类型的case clase（Person），然后将数据文件读入后隐式转换成SchemaRDD：people，并将people在sqlContext中注册成表rddTable，最后对表进行查询，找出年纪在13-19岁之间的人名。
/sparksql/people.txt的内容有3行：
运行下列代码：
case class Person(name:String,age:Int)
val rddpeople=sc.textFile(&/sparksql/people.txt&).map(_.split(&,&)).map(p=&Person(p(0),p(1).trim.toInt))
rddpeople.registerAsTable(&rddTable&)
sqlContext.sql(&SELECT name FROM rddTable WHERE age &= 13 AND age &= 19&).map(t =& &Name: & + t(0)).collect().foreach(println)运行结果：
1.2：parquet文件
同样得，sqlContext可以读取parquet文件，由于parquet文件中保留了schema的信息，所以不需要使用case class来隐式转换。sqlContext读入parquet文件后直接转换成SchemaRDD，也可以将SchemaRDD保存成parquet文件格式。
我们先将上面建立的SchemaRDD：people保存成parquet文件：
rddpeople.saveAsParquetFile(&/sparksql/people.parquet&)
运行后/sparksql/目录下就多出了一个名称为people.parquet的目录：
然后，将people.parquet读入，注册成表parquetTable，查询年纪大于25岁的人名：
//parquet演示
val parquetpeople = sqlContext.parquetFile(&/sparksql/people.parquet&)
parquetpeople.registerAsTable(&parquetTable&)
sqlContext.sql(&SELECT name FROM parquetTable WHERE age &= 25&).map(t =& &Name: & + t(0)).collect().foreach(println)运行结果：
1.3：json文件
sparkSQL1.1开始提供对json文件格式的支持，这意味着开发者可以使用更多的数据源，如鼎鼎大名的NOSQL数据库MongDB等。sqlContext可以从jsonFile或jsonRDD获取schema信息，来构建SchemaRDD，注册成表后就可以使用。
jsonFile - 加载JSON文件目录中的数据，文件的每一行是一个JSON对象。 jsonRdd - 从现有的RDD加载数据，其中RDD的每个元素包含一个JSON对象的字符串。
下面的例子读入一个json文件/sparksql/people.json，注册成jsonTable，并查询年纪大于25岁的人名。
/sparksql/people.json的内容：
运行下面代码：
//json演示
val jsonpeople = sqlContext.jsonFile(&/sparksql/people.json&)
jsonpeople.registerAsTable(&jsonTable&)
sqlContext.sql(&SELECT name FROM jsonTable WHERE age &= 25&).map(t =& &Name: & + t(0)).collect().foreach(println)运行结果：
&/pre&&img src=&http://img.blog.csdn.net/15849?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvYm9va19tbWlja3k=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast& alt=&& /&&/div&&div style=&line-height: 28 font-family: 'Hiragino Sans GB W3', 'Hiragino Sans GB', Arial, Helvetica, simsun, 宋体; font-size: 16&&2：hiveContext基础应用&/div&&div style=&line-height: 28 font-family: 'Hiragino Sans GB W3', 'Hiragino Sans GB', Arial, Helvetica, simsun, 宋体; font-size: 16&&      使用hiveContext之前首先要确认以下两点：&/div&&div style=&line-height: 28 font-family: 'Hiragino Sans GB W3', 'Hiragino Sans GB', Arial, Helvetica, simsun, 宋体; font-size: 16&&&ul style=&margin: 5px 0px 5px 40 padding: 0&&&li&使用的Spark是支持hive&/li&&li&hive的配置文件hive-site.xml已经存在conf目录中&/li&&/ul&      前者可以查看lib目录下是否存在以datanucleus开头的3个JAR来确定，后者注意是否在hive-site.xml里配置了uris来访问hive metastore。&/div&&div style=&line-height: 28 font-family: 'Hiragino Sans GB W3', 'Hiragino Sans GB', Arial, Helvetica, simsun, 宋体; font-size: 16&&&div&&/div&&div&要使用hiveContext，需要先构建hiveContext：&/div&&div&&pre name=&code& class=&html&&val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)然后就可以对hive数据进行操作了，下面我们将使用hive中的销售数据（第五小结中的hive数据），首先切换数据库到saledata并查看有几个表：
hiveContext.sql(&use saledata&)
hiveContext.sql(&show tables&).collect().foreach(println)可以看到有在第五小节定义的3个表：
现在查询一下所有订单中每年的销售单数、销售总额：
//所有订单中每年的销售单数、销售总额
//三个表连接后以count(distinct a.ordernumber)计销售单数，sum(b.amount)计销售总额
hiveContext.sql(&select c.theyear,count(distinct a.ordernumber),sum(b.amount) from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber join tbldate c on a.dateid=c.dateid group by c.theyear order by c.theyear&).collect().foreach(println)运行结果：
再做一个稍微复杂点的查询，求出所有订单每年最大金额订单的销售额：
/************************
所有订单每年最大金额订单的销售额:
第一步，先求出每份订单的销售额以其发生时间
select a.dateid,a.ordernumber,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber group by a.dateid,a.ordernumber
第二步，以第一步的查询作为子表，和表tblDate连接，求出每年最大金额订单的销售额
select c.theyear,max(d.sumofamount) from tbldate c join (select a.dateid,a.ordernumber,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber group by a.dateid,a.ordernumber ) d
on c.dateid=d.dateid group by c.theyear sort by c.theyear
*************************/
hiveContext.sql(&select c.theyear,max(d.sumofamount) from tbldate c join (select a.dateid,a.ordernumber,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber group by a.dateid,a.ordernumber ) d
on c.dateid=d.dateid group by c.theyear sort by c.theyear&).collect().foreach(println)运行结果：
最后做一个更复杂的查询，求出所有订单中每年最畅销货品：
/************************
所有订单中每年最畅销货品：
第一步：求出每年每个货品的销售金额
select c.theyear,b.itemid,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber join tbldate c on a.dateid=c.dateid group by c.theyear,b.itemid
第二步：求出每年单品销售的最大金额
select d.theyear,max(d.sumofamount) as maxofamount from (select c.theyear,b.itemid,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber join tbldate c on a.dateid=c.dateid group by c.theyear,b.itemid) d group by d.theyear
第三步：求出每年与销售额最大相符的货品就是最畅销货品
select distinct
e.theyear,e.itemid,f.maxofamount from (select c.theyear,b.itemid,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber join tbldate c on a.dateid=c.dateid group by c.theyear,b.itemid) e join (select d.theyear,max(d.sumofamount) as maxofamount from (select c.theyear,b.itemid,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber join tbldate c on a.dateid=c.dateid group by c.theyear,b.itemid) d group by d.theyear) f on (e.theyear=f.theyear and e.sumofamount=f.maxofamount) order by e.theyear
*************************/
hiveContext.sql(&select distinct
e.theyear,e.itemid,f.maxofamount from (select c.theyear,b.itemid,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber join tbldate c on a.dateid=c.dateid group by c.theyear,b.itemid) e join (select d.theyear,max(d.sumofamount) as maxofamount from (select c.theyear,b.itemid,sum(b.amount) as sumofamount from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber join tbldate c on a.dateid=c.dateid group by c.theyear,b.itemid) d group by d.theyear) f on (e.theyear=f.theyear and e.sumofamount=f.maxofamount) order by e.theyear&).collect().foreach(println)运行结果：
3：混合使用
在sqlContext和hiveContext中各自来源于不同数据源的表可以混用，但是sqlContext和hiveContext之间目前尚不能混合使用。
sqlContext中混合使用：
//sqlContext中混合使用
//sqlContext中来自rdd的表rddTable和来自parquet文件的表parquetTable混合使用
sqlContext.sql(&select a.name,a.age,b.age from rddTable a join parquetTable b on a.name=b.name&).collect().foreach(println)运行结果：
hiveContext中混合使用：
//hiveContext中混合使用
//创建一个hiveTable，并将数据加载，注意people.txt第二列有空格，所以age取string类型
hiveContext.sql(&CREATE TABLE hiveTable(name string,age string) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' &)
hiveContext.sql(&LOAD DATA LOCAL INPATH '/home/mmicky/mboo/MyClass/doc/sparkSQL/data/people.txt' INTO TABLE hiveTable&)
//创建一个源自parquet文件的表parquetTable2，然后和hiveTable混合使用
hiveContext.parquetFile(&/sparksql/people.parquet&).registerAsTable(&parquetTable2&)
hiveContext.sql(&select a.name,a.age,b.age from hiveTable a join parquetTable2 b on a.name=b.name&).collect().foreach(println)运行结果：
但是sqlContext中定义的表不能和hiveContext中定义的表混合使用。
4：缓存之使用
sparkSQL的cache可以使用两种方法来实现：
cacheTable()方法CACHE TABLE命令
千万不要先使用cache SchemaRDD，然后registerAsTable ；使用RDD的cache()将使用原生态的cache，而不是针对SQL优化后的内存列存储。看看cacheTable的源代码：
在默认的情况下，内存列存储的压缩功能是关闭的，要使用压缩功能需要配置变量COMPRESS_CACHED。
在sqlContext里可以如下使用cache：
//sqlContext的cache使用
sqlContext.cacheTable(&rddTable&)
sqlContext.sql(&SELECT name FROM rddTable WHERE age &= 13 AND age &= 19&).map(t =& &Name: & + t(0)).collect().foreach(println)
sqlContext.sql(&CACHE TABLE parquetTable&)
sqlContext.sql(&SELECT name FROM parquetTable WHERE age &= 13 AND age &= 19&).map(t =& &Name: & + t(0)).collect().foreach(println)观察webUI，可以看到cache的信息。（注意cache是lazy的，要有action才会实现；uncache是eager的，可以立即实现）
使用如下命令可以取消cache：
sqlContext.uncacheTable(&rddTable&)
sqlContext.sql(&UNCACHE TABLE parquetTable&)同样的，在hiveContext也可以使用上面的方法cache或uncache。
5：DSL之使用
sparkSQL除了支持HiveQL和SQL-92语法外，还支持DSL（Domain Specific Language）。在DSL中，使用scala符号'+标示符表示基础表中的列，spark的execution engine会将这些标示符隐式转换成表达式。另外可以在API中找到很多DSL相关的方法，如where()、select()、limit()等等，详细资料可以查看catalyst模块中的dsl子模块，下面为其中定义几种常用方法：
关于DSL的使用，随便举个例子，结合DSL方法，很容易上手：
val teenagers_dsl = rddpeople.where('age &= 10).where('age &= 19).select('name)
teenagers_dsl.map(t =& &Name: & + t(0)).collect().foreach(println)
上面介绍了sparkSQL的基础应用，sparkSQL还在高速发展中，存在者不少缺陷，如：
scala2.10.4本身对case class有22列的限制，在使用RDD数据源的时候就会造成不方便；sqlContext中3个表不能同时join，需要两两join后再join一次；sqlContext中不能直接使用values插入数据；。。。
总的来说，hiveContext还是令人满意，sqlContext就有些差强人意了。另外，顺便提一句，在编写sqlContext应用程序的时候，case class要定义在object之外。
最近将在炼数成金开课Spark大数据快速计算平台（第三期），本资料为新课素材。
延伸阅读：
上一篇讲解了如何在ABP中使用RedisCache，虽然能够正...
本教程为 李华明 编著的iOS-Cocos2d游戏开发系列教程：教程涵盖关于i......
专题主要学习DirectX的初级编程入门学习，对Directx11的入门及初学者有......
&面向对象的JavaScript&这一说法多少有些冗余，因为JavaScript 语言本......
Windows7系统专题 无论是升级操作系统、资料备份、加强资料的安全及管......Spark SQL & Spark Hive编程开发， 并和Hive执行效率对比 – 过往记忆
欢迎关注Hadoop、Hive、Hbase、Flume等微信公共账号：iteblog_hadoop。
文章总数：655
浏览总数：6,955,712
评论：3729
分类目录：78 个
注册用户数：1385
最后更新：日
欢迎关注微信公共帐号：iteblog_hadoop
　　 SQL也公布了很久，今天写了个程序来看下 SQL、Spark 以及直接用执行的效率进行了对比。以上测试都是跑在YARN上。
　　首先我们来看看我的环境：
3台DataNode，2台NameNode，每台机器20G内存，24核
数据都是lzo格式的，共336个文件，338.6 G
无其他任务执行
如果想及时了解Spark、Hadoop或者Hbase相关的文章，欢迎关注微信公共帐号：iteblog_hadoop
三个测试都是执行
select count(*), host, module
from ewaplog
group by host, module
order by host,
下面我们先来看看Spark SQL核心的代码（关于Spark SQL的详细介绍请参见Spark官方文档，这里我就不介绍了。）：
* User: 过往记忆
* Date: 14-8-13
* Time: 下午23:16
* 本文地址：/archives/1090
* 过往记忆博客，专注于hadoop、hive、spark、shark、flume的技术博客，大量的干货
* 过往记忆博客微信公共帐号：iteblog_hadoop
public static class Entry implements Serializable {
public Entry(String host, String module, String content) {
this.host =
this.module =
this.content =
public String getHost() {
public void setHost(String host) {
this.host =
public String getModule() {
public void setModule(String module) {
this.module =
public String getContent() {
public void setContent(String content) {
this.content =
public String toString() {
return &[& + host + &\t& + module + &\t& + content + &]&;
JavaSparkContext ctx = ...
JavaSQLContext sqlCtx = ...
JavaRDD&Entry& stringJavaRDD = ctx.textFile(args[0]).map(
new Function&String, Entry&() {
public Entry call(String str) throws Exception {
String[] split = str.split(&\u0001&);
if (split.length & 3) {
return new Entry(&&, &&, &&);
return new Entry(split[0], split[1], split[2]);
JavaSchemaRDD schemaPeople = sqlCtx.applySchema(stringJavaRDD, Entry.class);
schemaPeople.registerAsTable(&entry&);
JavaSchemaRDD teenagers = sqlCtx.sql(&select count(*), host, module & +
&from entry & +
&group by host, module & +
&order by host, module&);
List&String& teenagerNames = teenagers.map(new Function&Row, String&() {
public String call(Row row) {
return row.getLong(0) + &\t& +
row.getString(1) + &\t& + row.getString(2);
}).collect();
for (String name : teenagerNames) {
System.out.println(name);
Spark 核心代码：
* User: 过往记忆
* Date: 14-8-23
* Time: 下午23:16
* 本文地址：/archives/1090
* 过往记忆博客，专注于hadoop、hive、spark、shark、flume的技术博客，大量的干货
* 过往记忆博客微信公共帐号：iteblog_hadoop
JavaHiveContext hiveContext =....;
JavaSchemaRDD result = hiveContext.hql(&select count(*), host, module & +
&from ewaplog & +
&group by host, module & +
&order by host, module&);
List&Row& collect = result.collect();
for (Row row : collect) {
System.out.println(row.get(0) + &\t& + row.get(1) + &\t& + row.get(2));
　　大家可以看到Spark Hive核心代码里面的SQL语句和直接在Hive上面执行一样，在执行这个代码的时候，需要确保ewaplog存在。而且在运行这个程序的时候需要依赖Hive的一些jar包，需要依赖Hive的元数据等信息。对Hive的依赖比较大。而Spark SQL直接读取lzo文件，并没有涉及到Hive，相比Spark Hive依赖性这方便很好。Spark SQL直接读取lzo文件，然后将数据存放在RDD中，applySchema方法将JavaRDD转换成JavaSchemaRDD，我们来看看文档是怎么来描述的
　　At the core of this component is a new type of RDD, SchemaRDD. SchemaRDDs are composed Row objects along with a schema that describes the data types of each column in the row. A SchemaRDD is similar to a table in a traditional relational database. A SchemaRDD can be created from an existing RDD, Parquet file, a JSON dataset, or by running HiveQL against data stored in Apache Hive.
转换成JavaSchemaRDD之后，我们可以用用registerAsTable将它注册到表中，之后就可以通过JavaSQLContext的sql方法来执行相应的sql语句了。
　　用Maven编译完上面的程序之后，放到Hadoop集群上面运行：
iteblog@Spark $ spark-submit --master yarn-cluster
--jars lib/spark-sql_2.10-1.0.0.jar
--class SparkSQLTest
--queue queue1
./spark-1.0-SNAPSHOT.jar
/home/wyp/test/*.lzo
分别经过了20分钟左右的时间，Spark SQL和Spark Hive都可以运行完，结果如下：
bokingserver1 CN1_hbase_android_client
bokingserver1 CN1_hbase_iphone_client
bokingserver2 CN1_hbase_android_client
bokingserver2 CN1_hbase_iphone_client
bokingserver3 CN1_hbase_android_client
bokingserver3 CN1_hbase_iphone_client
bokingserver4 CN1_hbase_android_client
bokingserver4 CN1_hbase_iphone_client
bokingserver5 CN1_hbase_android_client
bokingserver5 CN1_hbase_iphone_client
bokingserver6 CN1_hbase_android_client
bokingserver6 CN1_hbase_iphone_client
bokingserver7 CN1_hbase_android_client
bokingserver7 CN1_hbase_iphone_client
bokingserver8 CN1_hbase_android_client
bokingserver8 CN1_hbase_iphone_client
bokingserver9 CN1_hbase_android_client
bokingserver9 CN1_hbase_iphone_client
bokingserver10 CN1_hbase_android_client
bokingserver10 CN1_hbase_iphone_client
bokingserver11 CN1_hbase_android_client
bokingserver11 CN1_hbase_iphone_client
bokingserver12 CN1_hbase_android_client
bokingserver12 CN1_hbase_iphone_client
　　为了比较基于Spark的任务确实比基于Mapreduce的快，我特意用Hive执行了同样的任务，如下：
hive& select count(*), host, module from ewaplog
& group by host, module order by host,
Job 0: Map: 2845
Reduce: 364
Cumulative CPU: 17144.59 sec
HDFS Read:
HDFS Write: 36516 SUCCESS
Job 1: Map: 1
Cumulative CPU: 4.82 sec
HDFS Read: 114193 HDFS Write: 1260 SUCCESS
Total MapReduce CPU Time Spent: 0 days 4 hours 45 minutes 49 seconds 410 msec
bokingserver1 CN1_hbase_android_client
bokingserver1 CN1_hbase_iphone_client
bokingserver2 CN1_hbase_android_client
bokingserver2 CN1_hbase_iphone_client
bokingserver3 CN1_hbase_android_client
bokingserver3 CN1_hbase_iphone_client
bokingserver4 CN1_hbase_android_client
bokingserver4 CN1_hbase_iphone_client
bokingserver5 CN1_hbase_android_client
bokingserver5 CN1_hbase_iphone_client
bokingserver6 CN1_hbase_android_client
bokingserver6 CN1_hbase_iphone_client
bokingserver7 CN1_hbase_android_client
bokingserver7 CN1_hbase_iphone_client
bokingserver8 CN1_hbase_android_client
bokingserver8 CN1_hbase_iphone_client
bokingserver9 CN1_hbase_android_client
bokingserver9 CN1_hbase_iphone_client
bokingserver10 CN1_hbase_android_client
bokingserver10 CN1_hbase_iphone_client
bokingserver11 CN1_hbase_android_client
bokingserver11 CN1_hbase_iphone_client
bokingserver12 CN1_hbase_android_client
bokingserver12 CN1_hbase_iphone_client
Time taken:
seconds, Fetched: 24 row(s)
　　从上面的显示我们可以看出，Hive执行同样的任务用了30分钟，而Spark用了20分钟，也就是省了1/3的时间，还是很快的。在运行的过程中，我发现Spark消耗内存比较大，在程序运行期间，三个子节点负载很高，整个队列的资源消耗了一半以上。我想如果集群的机器数据更多的话，Spark的运行速度应该还会有一些提升。好了今天就说到这，欢迎关注本博客。
本博客文章除特别声明，全部都是原创！
禁止个人和公司转载本文、谢谢理解：
下面文章您可能感兴趣你的浏览器禁用了JavaScript, 请开启后刷新浏览器获得更好的体验!
SparkSQL本质上是基于DAG模型的MPP。而Kylin核心是Cube(多维立方体)。关于MPP和Cube预处理的差异，重复如下：
& MPP [1] 的基本思路是增加机器来并行计算，从而提高查询速度。比如扫描8亿记录一台机器要处理1小时，但如果用100台机器来并行处理，就只要一分钟不到。再配合列式存储和一些索引，查询可以更快返回。要注意这里在线运算量并没有减小，8亿条记录还是要扫描一次，只是参与的机器多了，所以快了。
& MOLAP Cube [2][3] 是一种预计算技术，基本思路是预先对数据作多维索引，查询时只扫描索引而不访问原始数据从而提速。8亿记录的一个3维索引可能只有几万条记录，规模大大缩小，所以在线计算量大大减小，查询可以很快。索引表也可以采用列存储，并行扫描等MPP常用的技术。但多维索引要对多维度的各种组合作预计算，离线建索引需要较大计算量和时间，最终索引也会占用较多磁盘空间。
除了有无预处理的差异外，SparkSQL与Kylin对数据集大小的偏好也不一样。如果数据可以基本放入内存，Spark的内存缓存会让SparkSQL有好的表现。但对于超大规模的数据集，Spark也不能避免频繁的磁盘读写，性能会大幅下降。反过来Kylin的Cube预处理会大幅减小在线数据规模，对于超大规模数据更有优势。
Kylin作为YARN资源管理器一部分可以在Apache Hadoop的最新版本上获得。可能会慢一点，但比内存选项或那些不需要MapReduce处理的更具扩展性
要回复问题请先或
浏览: 1440
关注: 3 人}