Spark SQL是Spark中处理结构化数据的模块。与基础的Spark RDD API不同，Spark SQL的接口提供了更多关于数据的结构信息和计算任务的运行时信息。在Spark内部，Spark SQL会能够用于做优化的信息比RDD API更多一些。Spark SQL如今有了三种不同的API：SQL语句、DataFrame API和最新的Dataset API。不过真正运行计算的时候，无论你使用哪种API或语言，Spark SQL使用的执行引擎都是同一个。这种底层的统一，使开发者可以在不同的API之间来回切换，你可以选择一种最自然的方式，来表达你的需求。

本文中所有的示例都使用Spark发布版本中自带的示例数据，并且可以在spark-shell、pyspark shell以及sparkR shell中运行。

Spark SQL的一种用法是直接执行SQL查询语句，你可使用最基本的SQL语法，也可以选择HiveQL语法。Spark SQL可以从已有的Hive中读取数据。更详细的请参考Hive Tables 这一节。如果用其他编程语言运行SQL，Spark SQL将以DataFrame返回结果。你还可以通过命令行command-line 或者 JDBC/ODBC 使用Spark SQL。

DataFrame是一种分布式数据集合，每一条数据都由几个命名字段组成。概念上来说，她和关系型数据库的表 或者 R和Python中的data frame等价，只不过在底层，DataFrame采用了更多优化。DataFrame可以从很多数据源（sources）加载数据并构造得到，如：结构化数据文件，Hive中的表，外部数据库，或者已有的RDD。

DataFrame API支持Scala, Java, Python, and R。

Dataset是Spark-1.6新增的一种API，目前还是实验性的。Dataset想要把RDD的优势（强类型，可以使用lambda表达式函数）和Spark SQL的优化执行引擎的优势结合到一起。Dataset可以由JVM对象构建（constructed ）得到，而后Dataset上可以使用各种transformation算子（map，flatMap，filter 等）。

Dataset API 对 Scala 和 Java的支持接口是一致的，但目前还不支持Python，不过Python自身就有语言动态特性优势（例如，你可以使用字段名来访问数据，row.columnName）。对Python的完整支持在未来的版本会增加进来。

• Scala
• Java
• Python
• R

val sc: SparkContext // 假设已经有一个 SparkContext 对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 用于包含RDD到DataFrame隐式转换操作
import sqlContext.implicits._

除了SQLContext之外，你也可以创建HiveContext，HiveContext是SQLContext 的超集。

除了SQLContext的功能之外，HiveContext还提供了完整的HiveQL语法，UDF使用，以及对Hive表中数据的访问。要使用HiveContext，你并不需要安装Hive，而且SQLContext能用的数据源，HiveContext也一样能用。HiveContext是单独打包的，从而避免了在默认的Spark发布版本中包含所有的Hive依赖。如果这些依赖对你来说不是问题（不会造成依赖冲突等），建议你在Spark-1.3之前使用HiveContext。而后续的Spark版本，将会逐渐把SQLContext升级到和HiveContext功能差不多的状态。

spark.sql.dialect选项可以指定不同的SQL变种（或者叫SQL方言）。这个参数可以在SparkContext.setConf里指定，也可以通过 SQL语句的SET key=value命令指定。对于SQLContext，该配置目前唯一的可选值就是”sql”，这个变种使用一个Spark SQL自带的简易SQL解析器。而对于HiveContext，spark.sql.dialect 默认值为”hiveql”，当然你也可以将其值设回”sql”。仅就目前而言，HiveSQL解析器支持更加完整的SQL语法，所以大部分情况下，推荐使用HiveContext。

Spark应用可以用SparkContext创建DataFrame，所需的数据来源可以是已有的RDD（existing RDD），或者Hive表，或者其他数据源（data sources.）

以下是一个从JSON文件创建DataFrame的小栗子：

• Scala
• Java
• Python
• R

val sc: SparkContext // 已有的 SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

val df = sqlContext.read.json("examples/src/main/resources/people.json")

// 将DataFrame内容打印到stdout
df.show()

DataFrame提供了结构化数据的领域专用语言支持，包括Scala, Java, Python and R.

这里我们给出一个结构化数据处理的基本示例：

• Scala
• Java
• Python
• R

val sc: SparkContext // 已有的 SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 创建一个 DataFrame
val df = sqlContext.read.json("examples/src/main/resources/people.json")

// 展示 DataFrame 的内容
df.show()
// age  name
// null Michael
// 30   Andy
// 19   Justin

// 打印数据树形结构
df.printSchema()
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)

// select "name" 字段
df.select("name").show()
// name
// Michael
// Andy
// Justin

// 展示所有人，但所有人的 age 都加1
df.select(df("name"), df("age") + 1).show()
// name    (age + 1)
// Michael null
// Andy    31
// Justin  20

// 筛选出年龄大于21的人
df.filter(df("age") > 21).show()
// age name
// 30  Andy

// 计算各个年龄的人数
df.groupBy("age").count().show()
// age  count
// null 1
// 19   1
// 30   1

SQLContext.sql可以执行一个SQL查询，并返回DataFrame结果。

• Scala
• Java
• Python
• R

val sqlContext = ... // 已有一个 SQLContext 对象
val df = sqlContext.sql("SELECT * FROM table")

Dataset API和RDD类似，不过Dataset不使用Java序列化或者Kryo，而是使用专用的编码器（Encoder ）来序列化对象和跨网络传输通信。如果这个编码器和标准序列化都能把对象转字节，那么编码器就可以根据代码动态生成，并使用一种特殊数据格式，这种格式下的对象不需要反序列化回来，就能允许Spark进行操作，如过滤、排序、哈希等。

• Scala
• Java

// 对普通类型数据的Encoder是由 importing sqlContext.implicits._ 自动提供的
val ds = Seq(1, 2, 3).toDS()
ds.map(_ + 1).collect() // 返回: Array(2, 3, 4)

// 以下这行不仅定义了case class，同时也自动为其创建了Encoder
case class Person(name: String, age: Long)
val ds = Seq(Person("Andy", 32)).toDS()

// DataFrame 只需提供一个和数据schema对应的class即可转换为 Dataset。Spark会根据字段名进行映射。
val path = "examples/src/main/resources/people.json"
val people = sqlContext.read.json(path).as[Person]

Spark SQL有两种方法将RDD转为DataFrame。

1. 使用反射机制，推导包含指定类型对象RDD的schema。这种基于反射机制的方法使代码更简洁，而且如果你事先知道数据schema，推荐使用这种方式；

2. 编程方式构建一个schema，然后应用到指定RDD上。这种方式更啰嗦，但如果你事先不知道数据有哪些字段，或者数据schema是运行时读取进来的，那么你很可能需要用这种方式。

• Scala
• Java
• Python

// sc 是已有的 SparkContext 对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
// 为了支持RDD到DataFrame的隐式转换
import sqlContext.implicits._

// 定义一个case class.
// 注意：Scala 2.10的case class最多支持22个字段，要绕过这一限制，
// 你可以使用自定义class，并实现Product接口。当然，你也可以改用编程方式定义schema
case class Person(name: String, age: Int)

// 创建一个包含Person对象的RDD，并将其注册成table
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")

// sqlContext.sql方法可以直接执行SQL语句
val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")

// SQL查询的返回结果是一个DataFrame，且能够支持所有常见的RDD算子
// 查询结果中每行的字段可以按字段索引访问:
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

// 或者按字段名访问:
teenagers.map(t => "Name: " + t.getAs[String]("name")).collect().foreach(println)

// row.getValuesMap[T] 会一次性返回多列，并以Map[String, T]为返回结果类型
teenagers.map(_.getValuesMap[Any](List("name", "age"))).collect().foreach(println)
// 返回结果: Map("name" -> "Justin", "age" -> 19)

• Scala
• Java
• Python

// sc 是已有的SparkContext对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 创建一个RDD
val people = sc.textFile("examples/src/main/resources/people.txt")

// 数据的schema被编码与一个字符串中
val schemaString = "name age"

// Import Row.
import org.apache.spark.sql.Row;

// Import Spark SQL 各个数据类型
import org.apache.spark.sql.types.{StructType,StructField,StringType};

// 基于前面的字符串生成schema
val schema =
  StructType(
    schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))

// 将RDD[people]的各个记录转换为Rows，即：得到一个包含Row对象的RDD
val rowRDD = people.map(_.split(",")).map(p => Row(p(0), p(1).trim))

// 将schema应用到包含Row对象的RDD上，得到一个DataFrame
val peopleDataFrame = sqlContext.createDataFrame(rowRDD, schema)

// 将DataFrame注册为table
peopleDataFrame.registerTempTable("people")

// 执行SQL语句
val results = sqlContext.sql("SELECT name FROM people")

// SQL查询的结果是DataFrame，且能够支持所有常见的RDD算子
// 并且其字段可以以索引访问，也可以用字段名访问
results.map(t => "Name: " + t(0)).collect().foreach(println)

Spark SQL支持基于DataFrame操作一系列不同的数据源。DataFrame既可以当成一个普通RDD来操作，也可以将其注册成一个临时表来查询。把DataFrame注册为table之后，你就可以基于这个table执行SQL语句了。本节将描述加载和保存数据的一些通用方法，包含了不同的Spark数据源，然后深入介绍一下内建数据源可用选项。

在最简单的情况下，所有操作都会以默认类型数据源来加载数据（默认是Parquet，除非修改了spark.sql.sources.default 配置）。

• Scala
• Java
• Python
• R

val df = sqlContext.read.load("examples/src/main/resources/users.parquet")
df.select("name", "favorite_color").write.save("namesAndFavColors.parquet")

你也可以手动指定数据源，并设置一些额外的选项参数。数据源可由其全名指定（如，org.apache.spark.sql.parquet），而对于内建支持的数据源，可以使用简写名（json, parquet, jdbc）。任意类型数据源创建的DataFrame都可以用下面这种语法转成其他类型数据格式。

• Scala
• Java
• Python
• R

val df = sqlContext.read.format("json").load("examples/src/main/resources/people.json")
df.select("name", "age").write.format("parquet").save("namesAndAges.parquet")

Spark SQL还支持直接对文件使用SQL查询，不需要用read方法把文件加载进来。

• Scala
• Java
• Python
• R

val df = sqlContext.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`")

Save操作有一个可选参数SaveMode，用这个参数可以指定如何处理数据已经存在的情况。很重要的一点是，这些保存模式都没有加锁，所以其操作也不是原子性的。另外，如果使用Overwrite模式，实际操作是，先删除数据，再写新数据。

在使用HiveContext的时候，DataFrame可以用saveAsTable方法，将数据保存成持久化的表。与registerTempTable不同，saveAsTable会将DataFrame的实际数据内容保存下来，并且在HiveMetastore中创建一个游标指针。持久化的表会一直保留，即使Spark程序重启也没有影响，只要你连接到同一个metastore就可以读取其数据。读取持久化表时，只需要用用表名作为参数，调用SQLContext.table方法即可得到对应DataFrame。

默认情况下，saveAsTable会创建一个”managed table“，也就是说这个表数据的位置是由metastore控制的。同样，如果删除表，其数据也会同步删除。

Parquet 是一种流行的列式存储格式。Spark SQL提供对Parquet文件的读写支持，而且Parquet文件能够自动保存原始数据的schema。写Parquet文件的时候，所有的字段都会自动转成nullable，以便向后兼容。

仍然使用上面例子中的数据：

• Scala
• Java
• Python
• R
• Sql

// 我们继续沿用之前例子中的sqlContext对象
// 为了支持RDD隐式转成DataFrame
import sqlContext.implicits._

val people: RDD[Person] = ... // 和上面例子中相同，一个包含case class对象的RDD

// 该RDD将隐式转成DataFrame，然后保存为parquet文件
people.write.parquet("people.parquet")

// 读取上面保存的Parquet文件(多个文件 - Parquet保存完其实是很多个文件)。Parquet文件是自描述的，文件中保存了schema信息
// 加载Parquet文件，并返回DataFrame结果
val parquetFile = sqlContext.read.parquet("people.parquet")

// Parquet文件（多个）可以注册为临时表，然后在SQL语句中直接查询
parquetFile.registerTempTable("parquetFile")
val teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

像Hive这样的系统，一个很常用的优化手段就是表分区。在一个支持分区的表中，数据是保存在不同的目录中的，并且将分区键以编码方式保存在各个分区目录路径中。Parquet数据源现在也支持自动发现和推导分区信息。例如，我们可以把之前用的人口数据存到一个分区表中，其目录结构如下所示，其中有2个额外的字段，gender和country，作为分区键：

path
└── to
    └── table
        ├── gender=male
        │   ├── ...
        │   │
        │   ├── country=US
        │   │   └── data.parquet
        │   ├── country=CN
        │   │   └── data.parquet
        │   └── ...
        └── gender=female
            ├── ...
            │
            ├── country=US
            │   └── data.parquet
            ├── country=CN
            │   └── data.parquet
            └── ...

在这个例子中，如果需要读取Parquet文件数据，我们只需要把 path/to/table 作为参数传给 SQLContext.read.parquet 或者 SQLContext.read.load。Spark SQL能够自动的从路径中提取出分区信息，随后返回的DataFrame的schema如下：

root
|-- name: string (nullable = true)
|-- age: long (nullable = true)
|-- gender: string (nullable = true)
|-- country: string (nullable = true)

注意，分区键的数据类型将是自动推导出来的。目前，只支持数值类型和字符串类型数据作为分区键。

有的用户可能不想要自动推导出来的分区键数据类型。这种情况下，你可以通过 spark.sql.sources.partitionColumnTypeInference.enabled （默认是true）来禁用分区键类型推导。禁用之后，分区键总是被当成字符串类型。

从Spark-1.6.0开始，分区发现默认只在指定目录的子目录中进行。以上面的例子来说，如果用户把 path/to/table/gender=male 作为参数传给 SQLContext.read.parquet 或者 SQLContext.read.load，那么gender就不会被作为分区键。如果用户想要指定分区发现的基础目录，可以通过basePath选项指定。例如，如果把 path/to/table/gender=male作为数据目录，并且将basePath设为 path/to/table，那么gender仍然会最为分区键。

像ProtoBuffer、Avro和Thrift一样，Parquet也支持schema演变。用户从一个简单的schema开始，逐渐增加所需的新字段。这样的话，用户最终会得到多个schema不同但互相兼容的Parquet文件。目前，Parquet数据源已经支持自动检测这种情况，并合并所有文件的schema。

因为schema合并相对代价比较大，并且在多数情况下不是必要的，所以从Spark-1.5.0之后，默认是被禁用的。你可以这样启用这一功能：

• Scala
• Python
• R

// 继续沿用之前的sqlContext对象
// 为了支持RDD隐式转换为DataFrame
import sqlContext.implicits._

// 创建一个简单的DataFrame，存到一个分区目录中
val df1 = sc.makeRDD(1 to 5).map(i => (i, i * 2)).toDF("single", "double")
df1.write.parquet("data/test_table/key=1")

// 创建另一个DataFrame放到新的分区目录中，
// 并增加一个新字段，丢弃一个老字段
val df2 = sc.makeRDD(6 to 10).map(i => (i, i * 3)).toDF("single", "triple")
df2.write.parquet("data/test_table/key=2")

// 读取分区表
val df3 = sqlContext.read.option("mergeSchema", "true").parquet("data/test_table")
df3.printSchema()

// 最终的schema将由3个字段组成（single，double，triple）
// 并且分区键出现在目录路径中
// root
// |-- single: int (nullable = true)
// |-- double: int (nullable = true)
// |-- triple: int (nullable = true)
// |-- key : int (nullable = true)

在读写Hive metastore Parquet 表时，Spark SQL用的是内部的Parquet支持库，而不是Hive SerDe，因为这样性能更好。这一行为是由spark.sql.hive.convertMetastoreParquet 配置项来控制的，而且默认是启用的。

Hive和Parquet在表结构处理上主要有2个不同点：

由于以上原因，我们必须在Hive metastore Parquet table转Spark SQL Parquet table的时候，对Hive metastore schema做调整，调整规则如下：

Spark SQL会缓存Parquet元数据以提高性能。如果Hive metastore Parquet table转换被启用的话，那么转换过来的schema也会被缓存。这时候，如果这些表由Hive或其他外部工具更新了，你必须手动刷新元数据。

• Scala
• Java
• Python
• Sql

// 注意，这里sqlContext是一个HiveContext
sqlContext.refreshTable("my_table")

Parquet配置可以通过 SQLContext.setConf 或者 SQL语句中 SET key=value来指定。

• Scala
• Java
• Python
• R
• Sql

// sc是已有的SparkContext对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 数据集是由路径指定的
// 路径既可以是单个文件，也可以还是存储文本文件的目录
val path = "examples/src/main/resources/people.json"
val people = sqlContext.read.json(path)

// 推导出来的schema，可由printSchema打印出来
people.printSchema()
// root
//  |-- age: integer (nullable = true)
//  |-- name: string (nullable = true)

// 将DataFrame注册为table
people.registerTempTable("people")

// 跑SQL语句吧！
val teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")

// 另一种方法是，用一个包含JSON字符串的RDD来创建DataFrame
val anotherPeopleRDD = sc.parallelize(
  """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)
val anotherPeople = sqlContext.read.json(anotherPeopleRDD)

Spark SQL支持从Apache Hive读写数据。然而，Hive依赖项太多，所以没有把Hive包含在默认的Spark发布包里。要支持Hive，需要在编译spark的时候增加-Phive和-Phive-thriftserver标志。这样编译打包的时候将会把Hive也包含进来。注意，hive的jar包也必须出现在所有的worker节点上，访问Hive数据时候会用到（如：使用hive的序列化和反序列化SerDes时）。

Hive配置在conf/目录下hive-site.xml，core-site.xml（安全配置），hdfs-site.xml（HDFS配置）文件中。请注意，如果在YARN cluster（yarn-cluster mode）模式下执行一个查询的话，lib_mananged/jar/下面的datanucleus 的jar包，和conf/下的hive-site.xml必须在驱动器（driver）和所有执行器（executor）都可用。一种简便的方法是，通过spark-submit命令的–jars和–file选项来提交这些文件。

• Scala
• Java
• Python
• R

// sc是一个已有的SparkContext对象
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)

sqlContext.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
sqlContext.sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")

// 这里用的是HiveQL
sqlContext.sql("FROM src SELECT key, value").collect().foreach(println)

Spark SQL对Hive最重要的支持之一就是和Hive metastore进行交互，这使得Spark SQL可以访问Hive表的元数据。从Spark-1.4.0开始，Spark SQL有专门单独的二进制build版本，可以用来访问不同版本的Hive metastore，其配置表如下。注意，不管所访问的hive是什么版本，Spark SQL内部都是以Hive 1.2.1编译的，而且内部使用的Hive类也是基于这个版本（serdes，UDFs，UDAFs等）

以下选项可用来配置Hive版本以便访问其元数据：

Spark SQL也可以用JDBC访问其他数据库。这一功能应该优先于使用JdbcRDD。因为它返回一个DataFrame，而DataFrame在Spark SQL中操作更简单，且更容易和来自其他数据源的数据进行交互关联。JDBC数据源在java和python中用起来也很简单，不需要用户提供额外的ClassTag。（注意，这与Spark SQL JDBC server不同，Spark SQL JDBC server允许其他应用执行Spark SQL查询）

首先，你需要在spark classpath中包含对应数据库的JDBC driver，下面这行包括了用于访问postgres的数据库driver

SPARK_CLASSPATH=postgresql-9.3-1102-jdbc41.jar bin/spark-shell

远程数据库的表可以通过Data Sources API，用DataFrame或者SparkSQL 临时表来装载。以下是选项列表：

• Scala
• Java
• Python
• R
• Sql

val jdbcDF = sqlContext.read.format("jdbc").options(
  Map("url" -> "jdbc:postgresql:dbserver",
  "dbtable" -> "schema.tablename")).load()

• JDBC driver class必须在所有client session或者executor上，对java的原生classloader可见。这是因为Java的DriverManager在打开一个连接之前，会做安全检查，并忽略所有对原声classloader不可见的driver。最简单的一种方法，就是在所有worker节点上修改compute_classpath.sh，并包含你所需的driver jar包。
• 一些数据库，如H2，会把所有的名字转大写。对于这些数据库，在Spark SQL中必须也使用大写。

对于有一定计算量的Spark作业来说，可能的性能改进的方式，不是把数据缓存在内存里，就是调整一些开销较大的选项参数。

Spark SQL可以通过调用SQLContext.cacheTable(“tableName”)或者DataFrame.cache()把tables以列存储格式缓存到内存中。随后，Spark SQL将会扫描必要的列，并自动调整压缩比例，以减少内存占用和GC压力。你也可以用SQLContext.uncacheTable(“tableName”)来删除内存中的table。

你还可以使用SQLContext.setConf 或在SQL语句中运行SET key=value命令，来配置内存中的缓存。

以下选项同样也可以用来给查询任务调性能。不过这些选项在未来可能被放弃，因为spark将支持越来越多的自动优化。

Spark SQL可以作为JDBC/ODBC或者命令行工具的分布式查询引擎。在这种模式下，终端用户或应用程序，无需写任何代码，就可以直接在Spark SQL中运行SQL查询。

这里实现的Thrift JDBC/ODBC server和Hive-1.2.1中的HiveServer2是相同的。你可以使用beeline脚本来测试Spark或者Hive-1.2.1的JDBC server。

在Spark目录下运行下面这个命令，启动一个JDBC/ODBC server

./sbin/start-thriftserver.sh

这个脚本能接受所有 bin/spark-submit 命令支持的选项参数，外加一个 –hiveconf 选项，来指定Hive属性。运行./sbin/start-thriftserver.sh –help可以查看完整的选项列表。默认情况下，启动的server将会在localhost:10000端口上监听。要改变监听主机名或端口，可以用以下环境变量：

export HIVE_SERVER2_THRIFT_PORT=<listening-port>
export HIVE_SERVER2_THRIFT_BIND_HOST=<listening-host>
./sbin/start-thriftserver.sh /
  --master <master-uri> /
  ...

或者Hive系统属性 来指定

./sbin/start-thriftserver.sh /
  --hiveconf hive.server2.thrift.port=<listening-port> /
  --hiveconf hive.server2.thrift.bind.host=<listening-host> /
  --master <master-uri>
  ...

接下来，你就可以开始在beeline中测试这个Thrift JDBC/ODBC server:

./bin/beeline

下面的指令，可以连接到一个JDBC/ODBC server

beeline> !connect jdbc:hive2://localhost:10000

可能需要输入用户名和密码。在非安全模式下，只要输入你本机的用户名和一个空密码即可。对于安全模式，请参考beeline documentation.

Hive的配置是在conf/目录下的hive-site.xml，core-site.xml，hdfs-site.xml中指定的。

你也可以在beeline的脚本中指定。

Thrift JDBC server也支持通过HTTP传输Thrift RPC消息。以下配置（在conf/hive-site.xml中）将启用HTTP模式：

hive.server2.transport.mode - Set this to value: http
hive.server2.thrift.http.port - HTTP port number fo listen on; default is 10001
hive.server2.http.endpoint - HTTP endpoint; default is cliservice

同样，在beeline中也可以用HTTP模式连接JDBC/ODBC server:

beeline> !connect jdbc:hive2://<host>:<port>/<database>?hive.server2.transport.mode=http;hive.server2.thrift.http.path=<http_endpoint>

Spark SQL CLI是一个很方便的工具，它可以用local mode运行hive metastore service，并且在命令行中执行输入的查询。注意Spark SQL CLI目前还不支持和Thrift JDBC server通信。

用如下命令，在spark目录下启动一个Spark SQL CLI

./bin/spark-sql

Hive配置在conf目录下hive-site.xml，core-site.xml，hdfs-site.xml中设置。你可以用这个命令查看完整的选项列表：./bin/spark-sql –help

• 从Spark-1.6.0起，默认Thrift server 将运行于多会话并存模式下（multi-session）。这意味着，每个JDBC/ODBC连接有其独立的SQL配置和临时函数注册表。table的缓存仍然是公用的。如果你更喜欢老的单会话模式，只需设置spark.sql.hive.thriftServer.singleSession为true即可。当然，你也可在spark-defaults.conf中设置，或者将其值传给start-thriftserver.sh –conf（如下）：

./sbin/start-thriftserver.sh /
     --conf spark.sql.hive.thriftServer.singleSession=true /
     ...

• Tungsten引擎现在默认是启用的，Tungsten是通过手动管理内存优化执行计划，同时也优化了表达式求值的代码生成。这两个特性都可以通过把spark.sql.tungsten.enabled设为false来禁用。
• Parquet schema merging默认不启用。需要启用的话，设置spark.sql.parquet.mergeSchema为true即可
• Python接口支持用点(.)来访问字段内嵌值，例如df[‘table.column.nestedField’]。但这也意味着，如果你的字段名包含点号(.)的话，你就必须用重音符来转义，如：table.`column.with.dots`.nested。
• 列式存储内存分区剪枝默认是启用的。要禁用，设置spark.sql.inMemoryColumarStorage.partitionPruning为false即可
• 不再支持无精度限制的decimal。Spark SQL现在强制最大精度为38位。对于BigDecimal对象，类型推导将会使用（38，18）精度的decimal类型。如果DDL中没有指明精度，默认使用的精度是（10，0）
• 时间戳精确到1us（微秒），而不是1ns（纳秒）
• 在“sql”这个SQL变种设置中，浮点数将被解析为decimal。HiveQL解析保持不变。
• 标准SQL/DataFrame函数均为小写，例如：sum vs SUM。
• 当推测任务被启用是，使用DirectOutputCommitter是不安全的，因此，DirectOutputCommitter在推测任务启用时，将被自动禁用，且忽略相关配置。
• JSON数据源不再自动加载其他程序产生的新文件（例如，不是Spark SQL插入到dataset中的文件）。对于一个JSON的持久化表（如：Hive metastore中保存的表），用户可以使用REFRESH TABLE这个SQL命令或者HiveContext.refreshTable来把新文件包括进来。

根据用户的反馈，我们提供了一个新的，更加流畅的API，用于数据读（SQLContext.read）写（DataFrame.write），同时老的API（如：SQLCOntext.parquetFile, SQLContext.jsonFile）将被废弃。

有关SQLContext.read和DataFrame.write的更详细信息，请参考API文档。

根据用户的反馈，我们改变了DataFrame.groupBy().agg()的默认行为，在返回的DataFrame结果中保留了分组字段。如果你想保持1.3中的行为，设置spark.sql.retainGroupColumns为false即可。

• Scala
• Java
• Python

// 在1.3.x中，如果要保留分组字段"department", 你必须显式的在agg聚合时包含这个字段
df.groupBy("department").agg($"department", max("age"), sum("expense"))

// 而在1.4+，分组字段"department"默认就会包含在返回的DataFrame中
df.groupBy("department").agg(max("age"), sum("expense"))

// 要回滚到1.3的行为（不包含分组字段），按如下设置即可：
sqlContext.setConf("spark.sql.retainGroupColumns", "false")

在Spark 1.3中，我们去掉了Spark SQL的”Alpha“标签，并清理了可用的API。从Spark 1.3起，Spark SQL将对1.x系列二进制兼容。这个兼容性保证不包括显式的标注为”unstable（如：DeveloperAPI或Experimental）“的API。

对于用户来说，Spark SQL 1.3最大的改动就是SchemaRDD改名为DataFrame。主要原因是，DataFrame不再直接由RDD派生，而是通过自己的实现提供RDD的功能。DataFrame只需要调用其rdd方法就能转成RDD。

在Scala中仍然有SchemaRDD，只不过这是DataFrame的一个别名，以便兼容一些现有代码。但仍然建议用户改用DataFrame。Java和Python用户就没这个福利了，他们必须改代码。

在Spark 1.3之前，有单独的java兼容类（JavaSQLContext和JavaSchemaRDD）及其在Scala API中的镜像。Spark 1.3中将Java API和Scala API统一。两种语言的用户都应该使用SQLContext和DataFrame。一般这些类中都会使用两种语言中都有的类型（如：Array取代各语言独有的集合）。有些情况下，没有通用的类型（例如：闭包或者maps），将会使用函数重载来解决这个问题。

另外，java特有的类型API被删除了。Scala和java用户都应该用org.apache.spark.sql.types来编程描述一个schema。

Spark 1.3之前的很多示例代码，都在开头用 import sqlContext._，这行将会导致所有的sqlContext的函数都被引入进来。因此，在Spark 1.3我们把RDDs到DataFrames的隐式转换隔离出来，单独放到SQLContext.implicits对象中。用户现在应该这样写：import sqlContext.implicits._

另外，隐式转换也支持由Product（如：case classes或tuples）组成的RDD，但需要调用一个toDF方法，而不是自动转换。

如果需要使用DSL（被DataFrame取代的API）中的方法，用户之前需要导入DSL（import org.apache.spark.sql.catalyst.dsl）， 而现在应该要导入 DataFrame API（import org.apache.spark.sql.functions._）

Spark 1.3删除了sql包中的DataType类型别名。现在，用户应该使用 org.apache.spark.sql.types中的类。

注册UDF的函数，不管是DataFrame，DSL或者SQL中用到的，都被挪到SQLContext.udf中。

• Scala
• Java

sqlContext.udf.register("strLen", (s: String) => s.length())

Python UDF注册保持不变。

在python中使用DataTypes，你需要先构造一个对象（如：StringType()），而不是引用一个单例。

用户可以通过如下命令，为JDBC客户端session设定一个Fair Scheduler pool。

SET spark.sql.thriftserver.scheduler.pool=accounting;

在Shark中，默认的reducer个数是1，并且由mapred.reduce.tasks设定。Spark SQL废弃了这个属性，改为 spark.sql.shuffle.partitions, 并且默认200，用户可通过如下SET命令来自定义：

SET spark.sql.shuffle.partitions=10;
SELECT page, count(*) c
FROM logs_last_month_cached
GROUP BY page ORDER BY c DESC LIMIT 10;

你也可以把这个属性放到hive-site.xml中来覆盖默认值。

目前，mapred.reduce.tasks属性仍然能被识别，并且自动转成spark.sql.shuffle.partitions

shark.cache表属性已经不存在了，并且以”_cached”结尾命名的表也不再会自动缓存。取而代之的是，CACHE TABLE和UNCACHE TABLE语句，用以显式的控制表的缓存：

CACHE TABLE logs_last_month;
UNCACHE TABLE logs_last_month;

注意：CACHE TABLE tbl 现在默认是饥饿模式，而非懒惰模式。再也不需要手动调用其他action来触发cache了！

从Spark-1.2.0开始，Spark SQL新提供了一个语句，让用户自己控制表缓存是否是懒惰模式

CACHE [LAZY] TABLE [AS SELECT] ...

以下几个缓存相关的特性不再支持：

• 用户定义分区级别的缓存逐出策略
• RDD 重加载
• 内存缓存直接写入策略

Spark SQL设计时考虑了和Hive metastore，SerDes以及UDF的兼容性。目前这些兼容性斗是基于Hive-1.2.1版本，并且Spark SQL可以连到不同版本的Hive metastore（从0.12.0到1.2.1，参考：http://spark.apache.org/docs/latest/sql-programming-guide.html#interacting-with-different-versions-of-hive-metastore）

Spark SQL Thrift JDBC server采用了”out of the box”（开箱即用）的设计，使用很方便，并兼容已有的Hive安装版本。你不需要修改已有的Hive metastore或者改变数据的位置，或者表分区。

Spark SQL 支持绝大部分Hive功能，如：

• Hive查询语句：
  • SELECT
  • GROUP BY
  • ORDER BY
  • CLUSTER BY
  • SORT BY
• 所有的Hive操作符：
  • Relational operators (=, ⇔, ==, <>, <, >, >=, <=, etc)
  • Arithmetic operators (+, -, *, /, %, etc)
  • Logical operators (AND, &&, OR, ||, etc)
  • Complex type constructors
  • Mathematical functions (sign, ln, cos, etc)
  • String functions (instr, length, printf, etc)
• 用户定义函数（UDF）
• 用户定义聚合函数（UDAF）
• 用户定义序列化、反序列化（SerDes）
• 窗口函数（Window functions）
• Joins
  • JOIN
  • {LEFT|RIGHT|FULL} OUTER JOIN
  • LEFT SEMI JOIN
  • CROSS JOIN
• Unions
• 查询子句
  • SELECT col FROM ( SELECT a + b AS col from t1) t2
• 采样
• 执行计划详细（Explain）
• 分区表，包括动态分区插入
• 视图
• 所有Hive DDL（data definition language）：
  • CREATE TABLE
  • CREATE TABLE AS SELECT
  • ALTER TABLE
• 绝大部分Hive数据类型：
  • TINYINT
  • SMALLINT
  • INT
  • BIGINT
  • BOOLEAN
  • FLOAT
  • DOUBLE
  • STRING
  • BINARY
  • TIMESTAMP
  • DATE
  • ARRAY<>
  • MAP<>
  • STRUCT<>

以下是目前不支持的Hive特性的列表。多数是不常用的。

不支持的Hive常见功能

• bucket表：butcket是Hive表的一个哈希分区

不支持的Hive高级功能

• UNION类操作
• 去重join
• 字段统计信息收集：Spark SQL不支持同步的字段统计收集

Hive输入、输出格式

• CLI文件格式：对于需要回显到CLI中的结果，Spark SQL仅支持TextOutputFormat。
• Hadoop archive — Hadoop归档

Hive优化

一些比较棘手的Hive优化目前还没有在Spark中提供。有一些（如索引）对应Spark SQL这种内存计算模型来说并不重要。另外一些，在Spark SQL未来的版本中会支持。

• 块级别位图索引和虚拟字段（用来建索引）
• 自动计算reducer个数（join和groupBy算子）：目前在Spark SQL中你需要这样控制混洗后（post-shuffle）并发程度：”SET spark.sql.shuffle.partitions=[num_tasks];”
• 元数据查询：只查询元数据的请求，Spark SQL仍需要启动任务来计算结果
• 数据倾斜标志：Spark SQL不会理会Hive中的数据倾斜标志
• STREAMTABLE join提示：Spark SQL里没有这玩艺儿
• 返回结果时合并小文件：如果返回的结果有很多小文件，Hive有个选项设置，来合并小文件，以避免超过HDFS的文件数额度限制。Spark SQL不支持这个。

Spark SQL和DataFrames支持如下数据类型：

• Numeric types（数值类型）
  • ByteType: 1字节长的有符号整型，范围：-128 到 127.
  • ShortType: 2字节长有符号整型，范围：-32768 到 32767.
  • IntegerType: 4字节有符号整型，范围：-2147483648 到 2147483647.
  • LongType: 8字节有符号整型，范围： -9223372036854775808 to 9223372036854775807.
  • FloatType: 4字节单精度浮点数。
  • DoubleType: 8字节双精度浮点数
  • DecimalType: 任意精度有符号带小数的数值。内部使用java.math.BigDecimal, BigDecimal包含任意精度的不缩放整型，和一个32位的缩放整型
• String type（字符串类型）
  • StringType: 字符串
• Binary type（二进制类型）
  • BinaryType: 字节序列
• Boolean type（布尔类型）
  • BooleanType: 布尔类型
• Datetime type（日期类型）
  • TimestampType: 表示包含年月日、时分秒等字段的日期
  • DateType: 表示包含年月日字段的日期
• Complex types（复杂类型）
  • ArrayType(elementType, containsNull)：数组类型，表达一系列的elementType类型的元素组成的序列，containsNull表示数组能否包含null值
  • MapType(keyType, valueType, valueContainsNull)：映射集合类型，表示一个键值对的集合。键的类型是keyType，值的类型则由valueType指定。对应MapType来说，键是不能为null的，而值能否为null则取决于valueContainsNull。
  • StructType(fields)：表示包含StructField序列的结构体。
    • StructField(name, datatype, nullable): 表示StructType中的一个字段，name是字段名，datatype是数据类型，nullable表示该字段是否可以为空

• Scala
• Java
• Python
• R

import  org.apache.spark.sql.types._

这是Not-a-Number的缩写，某些float或double类型不符合标准浮点数语义，需要对其特殊处理：

• NaN == NaN，即：NaN和NaN总是相等
• 在聚合函数中，所有NaN分到同一组
• NaN在join操作中可以当做一个普通的join key
• NaN在升序排序中排到最后，比任何其他数值都大