本内容主要来自当前Spark最新版2.1.0的官方文档sql-programming-guide，以及一些其他阅读时搜索找到的相关辅助资料。

在所有工作开始前，也就是在官网文档中也没有介绍的就是，在pom文件里面添加spark-sql的依赖。不知为何官网没有把这个写进去。不过google下就知道了。这里建议去Spark Project SQL这个maven官网中去找当前最新的依赖添加到pom中。

这是我当时新加的依赖关系

Overview

Spark SQL是Spark的一个模块，用于结构化的数据处理。不像基础的RDD的API，Spark SQL提供了更多关于数据结构和计算的信息。Spark SQL使用了这些额外的信息来执行额外的优化。有多种方式和Spark SQL进行交互：包括SQL以及Dataset的API。不管使用的是哪种API或者哪种语言，最终的执行本质上是一样的。这种统一性意味着开发者可以轻松切换并使用最喜欢的方式来表示一个指定的Transformation。

SQL

通过SQL查询语句来使用Spark SQL是一种大家喜闻乐见的使用方式。而且Spark SQL还可以用来在已经存在的Hive装置中读取数据，后面会有介绍如何配置使用这一功能。在其他编程语言中使用SQL时，这条语句的返回值是Dataset/DataFrame类型的。同时你可以直接在cmd中与SQL接口交互，或者通过JDBC/ODBC（下面也有对其的介绍）。

Datasets和DataFrames

这里有两个重要的概念Datasets和Dataframes。（其实当你在网上搜索这两个名词时你会看到很多旧版本的解释，这多少可以帮助你理解，不过最终你还是要看最新版本的使用。只有真正理解了这两个概念才算是稍微懂了一些Spark SQL吧）

Dataset是一个分布式的数据收集器。这是在Spark1.6之后新加的一个接口，兼顾了RDD的优点（强类型，可以使用功能强大的lambda）以及Spark SQL的执行器高效性的优点。Dataset可以创建为一个JVM的对象，并使用Transformation功能（map,flatMap,filter, etc.）。注意：Dataset这个API只在Scala和Java中有。虽然Python中不支持这个API，但是根据Python语言的动态特性，其实这个API的很多好处Python已经具有了（比如row.columnName就可以访问数据）。在R语言中也类似。

其实Dataset主要由两部分构成，一个是具体的数据结构类对象，另一个就是和这个类对应的Encoder（啥是Encoder下面会有介绍，也可以google其他相关资料再看看）。

图画的不错吧

DataFrame是一个organized into named columns的Dataset。它在概念上与关系型数据库的table或者R/Python语言的DataFrame相似，不过被底层平台优化了。DataFrame可以从广泛的数据源中构成，比如：结构化的数据文件、Hive的table、外部数据库和RDD。DataFrame在Scala、Java、Python和R中都支持，在Scala中为Dataset[Row]，在Java中为Dataset<Row>。

其实如果想要再进一步了解DataFrame，就必须要了解Row的定义。其中有一个非常重要的属性：schema，后面也会有进一步的介绍。

“从代码中来看，DataFrame更像是Dataset的一种特殊情况。事实上是这样吗？。。”

Getting Started

要先有一个SparkSession

所有Spark功能的入口是类SparkSession，创建一个基础的SparkSession使用SparkSession.builder()方法：

具体的config选项有非常多及其对应的参数值

创建DataFrame

有了SparkSession，application就可以从RDD、Hive table或者其他Spark数据源（Data Sources）中创建出一个DataFrame

//从一个json文件中读取数据生成一个DataFrame

Dataset<Row> df=spark.read().json("/home/paul/spark/spark-2.1.0-bin-hadoop2.7/examples/src/main/resources/people.json");

df.show();

df.printSchema();

对应df.show()和df.printSchema()

无类型的Dataset操作（也称为DataFrame操作）

正如上面提到的，DataFrame在Spark2.0之后的Sala和Java中以Dataset of RowS方式存在。所以DataFrame的操作被称为无类型的操作，与其他Dataset的强类型操作形成对比。下面列举几个简单的操作：

df.printSchema();

df.select("name").show();

df.select(col("name"),col("age").plus(1)).show();

df.filter(col("age").gt(21)).show();

df.groupBy("age").count().show();

完整的DataFrame操作类型列表API：Class Dataset<T>

代码中运行SQL语句

SparkSession的方法sql允许application在代码中运行SQL语句，并得到Dataset<Row>类型的返回值。

// Register the DataFrame as a SQL temporary viewdf.createOrReplaceTempView("people");

DatasetsqlDF=spark.sql("SELECT * FROM people");

Global Temporary View

上面使用的是一个在Session生命周期中的临时views在Spark SQL中。如果你想拥有一个临时的view不过可以在不同的Session中共享，而且在application的运行周期内可用，那么就需要创建一个全局的临时view。并记得使用的时候加上global_temp作为前缀，因为全局的临时view是绑定到系统保留的数据库global_temp上。

// Register the DataFrame as a global temporary viewdf.createGlobalTempView("people");

// Global temporary view is tied to a system preserved database 注意people前面的global_temp

global_temp`spark.sql("SELECT * FROM global_temp.people").show();

// Global temporary view is cross-session

spark.newSession().sql("SELECT * FROM global_temp.people").show();

创建Datasets

Datasets和RDDs比较类似，不同的地方在与RDD是使用Java serialization或者Kryo实现序列化和反序列化，而Datasets是使用Encoder来实现对象的序列化并在网络中传输。Encoder的动态特性使得Spark可以在执行filtering、sorting和hashing等许多操作时无需把字节反序列化为对象。

创建一个用户自定义类的Encoder需要使用Java的beans，比如之前介绍中的代码：Encoder<Person> personEncoder=Encoders.bean(Person.class);

除了上面在讲Dataset时的一个创建Person类的Dataset的代码实例，下面再介绍下基础类型的以及从json文件中读取的情况。

Integer类型无需调用Encoder.bean

从上面的代码可以看出来，Dataset<Person>和DataFame（也就是Dataset<Row>）的区别就是DF后面没有as(类的Encoder)

转变RDD为DataFrame

Spark SQL支持两种不同的方法将RDD转换为DataFrame。第一种方法是根据RDD对象的具体类型映射（Reflection）推导出schema，这种映射为基础的方法可以让代码非常简洁，不过前提是在写application的时候已经知道schema的内容。

第二种方法是通过显式的程序代码构造schema，然后将这个schema应用到RDD上最后转化为Dataset。虽然这种方法会让代码变得比较复杂，但是它能在不知道Dataset的列名称及其类型的时候使用，也就是在代码运行时读出列的数据和类型。

第一种方法，使用Reflection推导出Schema

其实这个Reflection就是Spark SQL支持的JavaBean自动将RDD转换为DataFrame。通过Reflection获得到的BeanInfo定义为table的schema。下面是一段代码示例：

定义出一个RDD，然后转成DataFrame

获取到DataFrame之后对于列的索引可以通过index和name两种方式

两种Get row columns结果的方法

第二种方法：代码显式的构造Schema

当JavaBean无法根据类的具体内容提前定义出DataFrame时（比如：数据记录的结构对于不同的用户有不同的理解和使用），需要以下三个步骤创建出DataFrame。

1、创建一个RDD<Row>类型的RDD

2、创建类型为StructType的schema，与Step1中的RDD结构相匹配

3、通过SparkSession的方法createDataFrame将schema应用到RDD上产生DataFrame

代码显式生成DataFrame的示例

数据源Data Sources

Spark SQL通过DataFrame接口支持多种数据源。DataFrame可以使用相关的transformation操作以及用于产生临时的view。下面将展示一些与数据源相关的常用方法。

通用的Load/Save函数

最简单的方式，默认的数据源（parquet，除非有另外的配置spark.sql.sources.default）被用于所有的操作。

下面代码展示了从parquet文件读取生成DataFrame，保存为parquet文件以及直接对文件使用SQL语句的方法。

read()和write()之后可以加format("xxx")来指定格式

Save模式

Save操作可以选择多种SaveMode，来指定对于已经存在的文件做如何处理。而且需要特别明白的是这些save模式并没有实现任何锁机制而且也不是原子操作。下面是具体的：

SaveMode的各个定义及其含义

在代码中加入SaveMode也很简单。如下所示：

两种方法，一个是mode(saveMode: org.apache.spark.sql.SaveMode)，另一种是mode(saveMode: scala.Predef.String)

Save成Persistent Tables

DataFrame可以保存成persistent table到Hive的metastore通过代码saveAsTable。而且不需要存在Hive的部署，因为Spark会创建一个默认的本地Hive metastore（使用Derby）。并且不像createOrReplaceTempView，saveAsTable实现了DataFrame的内容而且创建了一个指针指向Hive metastore的数据。只要application一直保持对这个metastore的使用，那么这个persistent table就会一直存在，即使Spark程序已经重启了。SparkSession需要使用方法table来给DataFrame使用的persistent table命名。

默认的saveAsTable会创建“managed table”，意味着本地的数据将会被metastore控制，当table被drop的时候Managed table会自动delete相关数据。

Parquet文件

Parquet是一个列格式而且用于多个数据处理系统中。Spark SQL提供支持对于Parquet文件的读写，也就是自动保存原始数据的schema。当读Parquet文件时，所有的列被自动转化为nullable因为兼容性的缘故。

下面是对Parquet文件的一段代码处理：

对于Parquet文件的读写

Partition Discovery

Table partitioning是一个通用的优化方法在很多系统中使用，比如Hive。在一个partitioned table中，数据根据partitioning列不同的取值，通常被保存到多个不同的包含列取值名字的目录中。Parquet数据源现在可以发现并自动推导partitioning信息。下面是一个例子，新增加两个partitioning列gender和country到原先的partitioned table中。

Partition table结构示意

然后通过传入path/to/table到SparkSession.read.parquet或者SparkSession.read.load，Spark SQL可以自动从paths提取出partitioning信息。比如对于上面的那个例子，现在DataFrame的schema变为：

新加入gender和country的schema

另外需要注意两点：1、对于partitioning列的自动推导包含数据类型的推导，目前支持数值型的类型以及字符串型的类型。可以通过配置spark.sql.sources.partitionColumnTypeInference.enabled来选择这个自动类型推导的开关。默认是true，如果将其关闭那么所有类型认为是string类型；2、对于传入的paths参数，对于上面这个例子推荐使用path/to/table/gender=male的父目录也就是path/to/table/，不然gender不会被认为是一个partitioning列。

Schema合并

类似ProtocolBuffer、Avro以及Thrift，Parquet也支持Schema的演进。用户可以从一个简单的schema开始逐渐加入其它需要的列。在这种情况下，用户需要手动合并多个不同但兼容的Parquet文件。Parquet数据源目前已经可以自动检测并合并这些文件。

不过因为schema合并是一个相当昂贵的操作并且不是在所有的情况中都那么必需，所以Spark在1.5.0之后就关闭了自动合并。可以通过下面两张方法手动配置打开。

1、设置数据源选项mergeSchema为true，当reading Parquet文件时，或者

2、设置全局SQL选项spark.sql.parquet.mergeSchema为true

下面代码是一个例子，创建了一个Square类型的DataFrame和一个Cube类型的DataFrame。在保存为Parquet文件时特意配置了一个列值选项。在各自都保存好后，再从他们的父目录中读取，可以看到生成一个新的合并的DataFrame：

一个是squares的数据一个是cubes的数据，现在通过key合并为一个了

Hive metastore Parquet table转换

当读写到Hive metastore Parquet table时，Spark SQL将会使用自己的Parquet而不是Hive的SerDes为了更好的性能。当然可以通过配置spark.sql.hive.convertMetastoreParquet来控制开关，默认是打开的。

Hive/Parquet Schema解冲突

有两个关键的不同在Hive和Parquet对table生成schema的处理中：

1、Hive是不区分大小写的，但是Parquet区分

2、Hive认为所有的列是nullable，在Parquet中这只是列的一个特性。

基于上面的两点不同，我们必须解决Hive metastore schema和Parquet schema的冲突在转换Hive metastore Parquet table到Spark SQL Parquet table时。解决冲突的规则如下：

reconciled就是解决后的

Metadata更新

当有外部的Hive metastore对table操作时，可以在代码中手动刷新来保持一致。

更新table

Parquet配置

下面对于Parquet的配置通过使用SparkSession的方法setConf或者通过在SQL语句中运行SET key=value

Parquet配置列表

JSON Datasets

Spark SQL可以自动推导出schema从JSON数据集中，并保存为Dataset<Row>。这个转换可以从RDD<String>或者JSON文件中使用SparkSession.read().json()完成。下面代码是这两种转换的例子：

一个是从JSON文件中转换，另一个是从JSON格式的字符串中转换

Hive Table

Spark SQL同样支持读写存储在Apache Hive的数据。然而因为Hive需要巨大的依赖，所以这些依赖没有包含在目前默认的Spark发布版本中（这也就意味着不安装Hive就肯定无法完成下面的代码示例）。如果Hive的依赖可以在classpath中找到，Spark会自动加载。需要注意的是这些Hive的依赖也必须出现在所有的worker节点上，因为为了访问Hive存储的数据，他们也必须使用Hive的serialization和deserialization库（SerDes）

配置Hive就是将hive-site.xml、core-site.xml（安全配置）和hdfs-site.xml（HDFS配置）放到$SPARK_HOME下的conf/中。

TODO：我目前没有安装Hive，先挖个坑，回头有时间肯定要跳的。再填这个文档的坑。

代码1 从Hive数据文件中读取：

TODO：一个坑

代码2 将已有的DataFrame与Hive产生的JOIN

TODO：记得回来填坑

和不同版本的Hive Metastore交互

Spark SQL的Hive支持功能中一个最重要的点就是和Hive metastore交互，这使得Spark SQL可以访问Hive table的matadata。从Spark 1.4.0起一个单独的Spark SQL库可以被用于查询不同版本的Hive metastore，具体的配置信息如下表所示。 Note that independent of the version of Hive that is being used to talk to the metastore, internally Spark SQL will compile against Hive 1.2.1 and use those classes for internal execution (serdes, UDFs, UDAFs, etc).

配置Hive的版本号，来检索matadata

JDBC到其他数据库

Spark SQL同样支持通过JDBC读取其他数据库的数据作为数据源。在函数中推荐使用jdbcRDD，这是因为作为结果返回一个DataFrame可以方便的在Spark SQL中处理并且与其他数据源合并。具体的内容见官网文档：JDBC To Other Databases

附上官网的代码：

回头需要实际动手操作下

性能调优

性能调优主要是将数据放入内存中操作。通过spark.cacheTable("tableName")或者dataFrame.cache()。使用spark.uncacheTable("tableName")来从内存中去除table。

在SparkSession中的配置

其他配置选项（不过不怎么推荐手动修改，可能在后续版本自动的自适应修改）：

其他配置，但不建议手动修改

分布式的SQL引擎

Spark SQL同样可以通过使用它的JDBC/ODBC或者command-line接口作为分布式的查询引擎。在这种模式下，终端用户和应用可以通过SQL查询与Spark SQL直接交互而不需要其他额外的代码。

Running the Thrift JDBC/ODBC server

内容见spark.apache.org/docs/latest/sql-programming-guide.html#running-the-thrift-jdbcodbc-server

Running the Spark SQL CLI

内容见spark.apache.org/docs/latest/sql-programming-guide.html#running-the-spark-sql-cli