标签：
Spark SQL是支持在Spark中使用Sql、HiveSql、Scala中的关系型查询表达式。它的核心组件是一个新增的RDD类型SchemaRDD，它把行对象用一个Schema来描述行里面的所有列的数据类型，它就像是关系型数据库里面的一张表。它可以从原有的RDD创建，也可以是Parquet文件，最重要的是它可以支持用HiveQL从hive里面读取数据。
下面是一些案例，可以在Spark shell当中运行。
首先我们要创建一个熟悉的Context，熟悉spark的人都知道吧，有了Context我们才可以进行各种操作。
val sc: SparkContext // 已经存在的SparkContext val sqlContext = new org.apache.spark.sql.SQLContext(sc) import sqlContext._
Spark SQL通过SchemaRDD接口支持在多种数据源上进行操作。一旦一个数据集被加载，它可以被注册成一个表格，甚至可以和其它数据源有连接。
Spark SQL支持的一种表的类型是Scala的case class，case class定义了表的类型，下面是例子：
// sc是一个已经存在的SprakContext
val sqlContext = new org.apache.spark.sql.SQLContext(sc) // import sqlContext._ import sqlContext.createSchemaRDD // case class在Scala 2.10里面最多支持22个列，为了突破这个限制，最好是定义一个类实现Product接口 case class Person(name: String, age: Int) // 为Person的对象创建一个RDD，然后注册成一张表 val people = sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)) people.registerAsTable("people") // 直接写sql吧，这个方法是sqlContext提供的 val teenagers = sql("SELECT name FROM people WHERE age >= 13 AND age <= 19") // teenagers是SchemaRDDs类型，它支持所有普通的RDD操作 teenagers.map(t => "Name: " + t(0)).collect().foreach(println)
从上面这个方法来看，不是很好用，一个表好几十个字段，我就得一个一个的去赋值，它现在支持的操作都是很简单的操作，想要实现复杂的操作可以具体去看HiveContext提供的HiveQL。
Parquet是一种列式存储格式并且被许多数据处理系统支持。Parquet为Hadoop生态系统中的所有项目提供支持高效率压缩的列式数据表达，而且与数据处理框架、数据模型或编程语言都没有关系。Spark SQL提供了对Parquet的读和写，自动保留原始数据的架构。
val sqlContext = new org.apache.spark.sql.SQLContext(sc) // import sqlContext._ // createSchemaRDD被用来将RDD隐式转换成一个SchemaRDD
import sqlContext.createSchemaRDD
val people: RDD[Person] = ... // 同上面的例子. // 这个RDD已经隐式转换成一个SchemaRDD, 允许它存储成Parquet格式. people.saveAsParquetFile("people.parquet") // 从上面创建的文件里面读取，加载一个Parquet文件的结果也是一种JavaSchemaRDD. val parquetFile = sqlContext.parquetFile("people.parquet") //注册成表，然后在SQL状态下使用 parquetFile.registerAsTable("parquetFile") val teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19") teenagers.map(t => "Name:" + t(0)).collect().foreach(println)
JSON(JavaScript Object Notation) 是一种轻量级的数据交换格式。它基于JavaScript（Standard ECMA-262 3rd Edition - December 1999）的一个子集。 JSON采用完全独立于语言的文本格式，但是也使用了类似于C语言家族的习惯（包括C, C++, C#, Java, JavaScript, Perl, Python等）。这些特性使JSON成为理想的数据交换语言。易于人阅读和编写，同时也易于机器解析和生成(网络传输速度快)。
SparkSQL可以自动推断出一个JSON数据集模式并作为一个SchemaRDD来加载。这种转换可以通过使用SQLContext中的两个方法中的一个得到：
jsonFile - 从JSON文件的目录中加载数据，其中文件的每一行就是一个JSON对象。
jsonRdd - 从一个已存在的RDD中加载数据，其中每一个RDD元素都是一个包含一个JSON对象的字符串。
// sc 是已经存在的SparkContext val sqlContext = new org.apache.spark.sql.SQLContext(sc) // 一个JSON 数据集用一个路径指出 // 这个路径既可以是一个单独的文本文件，也可以是一个存储文本文件的目录 val path = "examples/src/main/resources/people.json" // 根据路径指出的文件生成一个SchemaRDD val people = sqlContext.jsonFile(path) // 推断的模式可以通过使用printSchema() 方法显式化 people.printSchema() // root // |-- age: IntegerType // |-- name: StringType // 把SchemaRDD注册成一个表 people.registerAsTable("people") // SQL状态可以通过使用sqlContext提供的sql方法运行 val teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19") // 另外，一个SchemaRDD也可以通过每个字符串存储一个JSON数据集对象的string类型的RDD来生成 val anotherPeopleRDD = sc.parallelize( """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil) val anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD)
Spark SQL也支持读写存储在Apache Hive上的数据。然而，hive的依赖太多了，默认的Spark assembly 是没带这些依赖的，需要我们运行 SPARK_HIVE=true sbt/sbt assembly/assembly重新编译，或者用maven的时候添加 -Phive参数，它会重新编译出来一个hive assembly的jar包，然后需要把这个jar包放到所有的节点上。另外还需要把 hive-site.xml放到conf目录下。没进行hive部署的话，下面的例子也可以用LocalHiveContext来代替HiveContext。
val sc: SparkContext // 已经存在的SparkContext val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) // 引入这个Context，然后就会给所有的sql语句进行隐式转换 import hiveContext._ hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") hql("LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt‘ INTO TABLE src") // 使用HiveQL查询 hql("FROM src SELECT key, value").collect().foreach(println)
或者写成如下形式：
// sc is an existing SparkContext. val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) hiveContext.hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") hiveContext.hql("LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt‘ INTO TABLE src") // Queries are expressed in HiveQL hiveContext.hql("FROM src SELECT key, value").collect().foreach(println)
文字语言综合关联查询，目前这个功能只是在Scala里面支持。
Spark SQL还支持一个特定域的语言编写查询。再次，利用上述实例数据：
// sc是一个已经存在的SparkContext
val sqlContext = new org.apache.spark.sql.SQLContext(sc) import sqlContext._ val people: RDD[Person] = ... // 同前面的例子. // 和后面这个语句是一样的 ‘SELECT name FROM people WHERE age >= 10 AND age <= 19‘ val teenagers = people.where(‘age >= 10).where(‘age <= 19).select(‘name)
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)DSL（领域语言）使用scala符号表示隐含表中的列，通过在前面加一个（‘）来标示。隐式转换将这些符号表达式表示成SQL执行引擎的值。一个完整的功能支持列表可以在ScalaDoc中找到。
标签：
原文地址：http://www.cnblogs.com/gaopeng527/p/4315808.html
