Spark SQL
简单的说Spark SQL是spark用来操作结构化和半结构化数据的接口。本文来讲述一下它的一些基本操作。
Spark SQL的特性
- 无缝地将SQL查询和spark程序混合，与常规的Python/Java/scala代码高度整合，包含了连接RDD与SQL表、公开的自定义SQL函数接口等。
- 可以从各种结构化数据源中读取数据，如（JSON、HIVE等）
- 可以通过JDBC或者ODBC连接，Spark SQL包括有行业标准的JDBC和ODBC连接的服务器模式。
为了实现这些功能Spark SQL提供了一种特殊的RDD叫做SchemaRDD，这是一个存放row对象的RDD，每个RDD代表一行记录。
Spark SQL编译时可以包含hive支持（Apache Hive 是 Hadoop上的SQL引擎），也可以不包含，包含Hive支持的Spark SQL可以支持hive表访问、UDF（用户自定义函数）、SerDe（序列化格式和反序列化格式）、以及Hive查询语言（HiveQL/HQL）。
下面来说一下Spark SQL的操作。
初始化Spark SQL
Scala中SQL的import声明：
//导入 Spark SQL import org.apache.spark.sql.hive.HiveContext //如果不能使用hive依赖的话 import org.apahe.spark.sql.SQLContext
需要导入的隐式转换支持
//创建Spark SQL的HiveContext val hiveCtx = ... //如果不能使用hive依赖的话 import hiveCtx._
添加好import声明之后，需要创建出一个HiveContext对象，如果无法一如Hive依赖，就创建出一个SQLContext对象作为SQL的上下文环境，这两个类都要传入一个SparkContext对象作为运行的基础
创建SQL上下文环境
val sc = new SparkContext(...) val hiveCtx = new HiveContext(sc)
有了HiveContext或SQLContext后我们就可以准备读取数据并进行查询了。
基本查询示例
读取并查询推文
想要在一张数据表上进行查询，需要调用HiveContext或者SQLContext中的sql()方法，要做的第一件事就是告诉Spark SQL要查询的数据是什么。
val input = hiveCtx.jsonFile(inputFile) //注册输入的SchemaRDD input.registerTempTable("tweets") //依据retweetCount(转发计数)选出推文 val topTweets = hiveCtx.sql("SELECT text,retweetCount FROM tweet ORDER BY retweetCount LIMIT 10")
SchemaRDD
我们在读取和执行查询的时候都会返回SchemaRDD，SchemaRDD和传统的数据库中的表的概念类似。SchemaRDD是一个由ROW对象组成的RDD，附带包含每列数据类型的结构信息。
row对象
row对象表示SchemaRDD中的记录，其本质就是一个定长的字段数组，在Scala中Row对象有一系列Getter方法，可以通过下标来获取每个字段的值。
例：在Scala中访问topTweet这个SchemaRDD中的text列（第一列）
val topTweetText = topTweets.map(row => row.getString(0))
读取和存储数据
Spark SQL支持很多种结构化数据源，可以让你跳过复杂的读取过程，轻松从各种数据源中读取到Row对象，这些数据源包括Hive表、JSON和Parquet文件。
Hive
从hive中读取数据时，Spark SQL支持任何Hive支持的存储格式（SerDe），包括文本文件，RCFiles，ORC，Parquet，Avro以及Protocol Buffer。
例：如何查询一张hive表。
import org.apache.spark.sql.hive.HiveContext val hiveCtx = new HiveContext(sc) val rows = hiveCtx.sql("SELECT key,value Frow mytable") val keys = row.map(row => row.getInt(0))
Parquet
Parquet是一种流行的列式存储格式，可以高效地存储具有嵌套字段的记录，Parquet格式经常在Hadoop生态圈中被使用，他也支持Spark SQL的全部数据类型。
JSON
要读取JSON数据，只要调用hiveCtx中的jsonFile()方法即可，如果想获得从数据中推断出来的结构信息，可以在生成的SchemaRDD上调用printSchema方法。
在Scala中使用Spark SQL读取json数据。
val input = hiveCtx.jsonFile(inputFile)
基于RDD创建SchemaRDD
case class Happyperson(handle:String,favouriteBeverage:String) ... //创建了一个人的对象，并且把它转换成SchemaRDD val HappypersonRDD = sc.parallelize(List(Happyperson("holden","coffee"))) happyPeopleRDD.registerTempTable("happy_people")
小结
本文讲解了Spark利用Spark SQL进行结构化和半结构化数据处理的方式，除了上述的一些方法，操作RDD的方法也同样适用于SchemaRDD。
