Spark SQL, DataFrames and Datasets Guide
无类型的Dataset操作 (aka DataFrame 操作)
Running SQL Queries Programmatically
Untyped User-Defined Aggregate Functions
Type-Safe User-Defined Aggregate Functions
Generic Load/Save Functions （通用 加载/保存 功能）
Manually Specifying Options （手动指定选项）
Run SQL on files directly （直接在文件上运行 SQL）
Saving to Persistent Tables （保存到持久表）
Bucketing, Sorting and Partitioning （分桶, 排序和分区）
Loading Data Programmatically （以编程的方式加载数据）
Hive metastore Parquet table conversion （Hive metastore Parquet table 转换）
Hive/Parquet Schema Reconciliation
DataFrame data reader/writer interface
DataFrame.groupBy 保留 grouping columns（分组的列）
针对 DataType 删除在 org.apache.spark.sql 包中的一些类型别名（仅限于 Scala）
UDF 注册迁移到sqlContext.udf中 (Java & Scala)
Python DataTypes 不再是 Singletons（单例的）
Spark SQL 是 Spark 处理结构化数据的一个模块.与基础的 Spark RDD API 不同, Spark SQL 提供了查询结构化数据及计算结果等信息的接口.在内部, Spark SQL 使用这个额外的信息去执行额外的优化.有几种方式可以跟 Spark SQL 进行交互, 包括 SQL 和 Dataset API.当使用相同执行引擎进行计算时, 无论使用哪种 API / 语言都可以快速的计算.这种统一意味着开发人员能够在基于提供最自然的方式来表达一个给定的 transformation API 之间实现轻松的来回切换不同的 .
该页面所有例子使用的示例数据都包含在 Spark 的发布中, 并且可以使用spark-shell,pysparkshell, 或者sparkRshell来运行.
Spark SQL 的功能之一是执行 SQL 查询.Spark SQL 也能够被用于从已存在的 Hive 环境中读取数据.更多关于如何配置这个特性的信息, 请参考Hive 表这部分. 当以另外的编程语言运行SQL 时, 查询结果将以Dataset/DataFrame的形式返回.您也可以使用命令行或者通过JDBC/ODBC与 SQL 接口交互.
一个 Dataset 是一个分布式的数据集合 Dataset 是在 Spark 1.6 中被添加的新接口, 它提供了 RDD 的优点（强类型化, 能够使用强大的 lambda 函数）与Spark SQL执行引擎的优点.一个 Dataset 可以从 JVM 对象来构造并且使用转换功能（map, flatMap, filter, 等等）. Dataset API 在Scala和Java是可用的.Python 不支持 Dataset API.但是由于 Python 的动态特性, 许多 Dataset API 的优点已经可用了 (也就是说, 你可能通过 name 天生的row.columnName属性访问一行中的字段).这种情况和 R 相似.
一个 DataFrame 是一个Dataset组成的指定列.它的概念与一个在关系型数据库或者在 R/Python 中的表是相等的, 但是有很多优化. DataFrames 可以从大量的sources中构造出来, 比如: 结构化的文本文件, Hive中的表, 外部数据库, 或者已经存在的 RDDs. DataFrame API 可以在 Scala, Java,Python, 和R中实现. 在 Scala 和 Java中, 一个 DataFrame 所代表的是一个多个Row（行）的的 Dataset（数据集合）. 在the Scala API中,DataFrame仅仅是一个Dataset[Row]类型的别名. 然而, 在Java API中, 用户需要去使用Dataset去代表一个DataFrame.
在此文档中, 我们将常常会引用 Scala/Java Datasets 的Rows 作为 DataFrames.
Spark SQL中所有功能的入口点是SparkSession类. 要创建一个SparkSession, 仅使用SparkSession.builder()就可以了:
importorg.apache.spark.sql.SparkSessionvalspark=SparkSession.builder().appName("Spark SQL basic example").config("spark.some.config.option","some-value").getOrCreate()// For implicit conversions like converting RDDs to DataFramesimportspark.implicits._
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Spark 2.0 中的SparkSession为 Hive 特性提供了内嵌的支持, 包括使用 HiveQL 编写查询的能力, 访问 Hive UDF,以及从 Hive 表中读取数据的能力.为了使用这些特性, 你不需要去有一个已存在的 Hive 设置.
在一个SparkSession中, 应用程序可以从一个已经存在的RDD, 从hive表, 或者从Spark数据源中创建一个DataFrames.
举个例子, 下面就是基于一个JSON文件创建一个DataFrame:
valdf=spark.read.json("examples/src/main/resources/people.json")// Displays the content of the DataFrame to stdoutdf.show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
无类型的Dataset操作 (aka DataFrame 操作)
DataFrames 提供了一个特定的语法用在Scala,Java,PythonandR中机构化数据的操作.
正如上面提到的一样, Spark 2.0中, DataFrames在Scala 和 Java API中, 仅仅是多个Rows的Dataset. 这些操作也参考了与强类型的Scala/Java Datasets中的”类型转换” 对应的”无类型转换” .
这里包括一些使用 Dataset 进行结构化数据处理的示例 :
// This import is needed to use the $-notationimportspark.implicits._// Print the schema in a tree formatdf.printSchema()// root// |-- age: long (nullable = true)// |-- name: string (nullable = true)// Select only the "name" columndf.select("name").show()// +-------+// | name|// +-------+// |Michael|// | Andy|// | Justin|// +-------+// Select everybody, but increment the age by 1df.select($"name",$"age"+1).show()// +-------+---------+// | name|(age + 1)|// +-------+---------+// |Michael| null|// | Andy| 31|// | Justin| 20|// +-------+---------+// Select people older than 21df.filter($"age">21).show()// +---+----+// |age|name|// +---+----+// | 30|Andy|// +---+----+// Count people by agedf.groupBy("age").count().show()// +----+-----+// | age|count|// +----+-----+// | 19| 1|// |null| 1|// | 30| 1|// +----+-----+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
能够在 DataFrame 上被执行的操作类型的完整列表请参考API 文档.
除了简单的列引用和表达式之外, DataFrame 也有丰富的函数库, 包括 string 操作, date 算术, 常见的 math 操作以及更多.可用的完整列表请参考DataFrame 函数指南.
Running SQL Queries Programmatically
SparkSession的sql函数可以让应用程序以编程的方式运行 SQL 查询, 并将结果作为一个DataFrame返回.
// Register the DataFrame as a SQL temporary viewdf.createOrReplaceTempView("people")valsqlDF=spark.sql("SELECT * FROM people")sqlDF.show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Spark SQL中的临时视图是session级别的, 也就是会随着session的消失而消失. 如果你想让一个临时视图在所有session中相互传递并且可用, 直到Spark 应用退出, 你可以建立一个全局的临时视图.全局的临时视图存在于系统数据库global_temp中, 我们必须加上库名去引用它, 比如.SELECT * FROM global_temp.view1.
// Register the DataFrame as a global temporary viewdf.createGlobalTempView("people")// Global temporary view is tied to a system preserved database `global_temp`spark.sql("SELECT * FROM global_temp.people").show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+// Global temporary view is cross-sessionspark.newSession().sql("SELECT * FROM global_temp.people").show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Dataset 与 RDD 相似, 然而, 并不是使用 Java 序列化或者 Kryo编码器来序列化用于处理或者通过网络进行传输的对象. 虽然编码器和标准的序列化都负责将一个对象序列化成字节, 编码器是动态生成的代码, 并且使用了一种允许 Spark 去执行许多像 filtering, sorting 以及 hashing 这样的操作, 不需要将字节反序列化成对象的格式.
// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,// you can use custom classes that implement the Product interfacecaseclassPerson(name:String,age:Long)// Encoders are created for case classesvalcaseClassDS=Seq(Person("Andy",32)).toDS()caseClassDS.show()// +----+---+// |name|age|// +----+---+// |Andy| 32|// +----+---+// Encoders for most common types are automatically provided by importing spark.implicits._valprimitiveDS=Seq(1,2,3).toDS()primitiveDS.map(_+1).collect()// Returns: Array(2, 3, 4)// DataFrames can be converted to a Dataset by providing a class. Mapping will be done by namevalpath="examples/src/main/resources/people.json"valpeopleDS=spark.read.json(path).as[Person]peopleDS.show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Spark SQL 支持两种不同的方法用于转换已存在的 RDD 成为 Dataset.第一种方法是使用反射去推断一个包含指定的对象类型的 RDD 的 Schema.在你的 Spark 应用程序中当你已知 Schema 时这个基于方法的反射可以让你的代码更简洁.
第二种用于创建 Dataset 的方法是通过一个允许你构造一个 Schema 然后把它应用到一个已存在的 RDD 的编程接口.然而这种方法更繁琐, 当列和它们的类型知道运行时都是未知时它允许你去构造 Dataset.
Spark SQL 的 Scala 接口支持自动转换一个包含 case classes 的 RDD 为 DataFrame.Case class 定义了表的 Schema.Case class 的参数名使用反射读取并且成为了列名.Case class 也可以是嵌套的或者包含像Seq或者Array这样的复杂类型.这个 RDD 能够被隐式转换成一个 DataFrame 然后被注册为一个表.表可以用于后续的 SQL 语句.
// For implicit conversions from RDDs to DataFramesimportspark.implicits._// Create an RDD of Person objects from a text file, convert it to a DataframevalpeopleDF=spark.sparkContext.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(attributes=>Person(attributes(0),attributes(1).trim.toInt)).toDF()// Register the DataFrame as a temporary viewpeopleDF.createOrReplaceTempView("people")// SQL statements can be run by using the sql methods provided by SparkvalteenagersDF=spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")// The columns of a row in the result can be accessed by field indexteenagersDF.map(teenager=>"Name: "+teenager(0)).show()// +------------+// | value|// +------------+// |Name: Justin|// +------------+// or by field nameteenagersDF.map(teenager=>"Name: "+teenager.getAs[String]("name")).show()// +------------+// | value|// +------------+// |Name: Justin|// +------------+// No pre-defined encoders for Dataset[Map[K,V]], define explicitlyimplicitvalmapEncoder=org.apache.spark.sql.Encoders.kryo[Map[String,Any]]// Primitive types and case classes can be also defined as// implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder()// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]teenagersDF.map(teenager=>teenager.getValuesMap[Any](List("name","age"))).collect()// Array(Map("name" -> "Justin", "age" -> 19))
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
当 case class 不能够在执行之前被定义（例如, records 记录的结构在一个 string 字符串中被编码了, 或者一个 text 文本 dataset 将被解析并且不同的用户投影的字段是不一样的）.一个DataFrame可以使用下面的三步以编程的方式来创建.
从原始的 RDD 创建 RDD 的Row（行）;
Step 1 被创建后, 创建 Schema 表示一个StructType匹配 RDD 中的Row（行）的结构.
通过SparkSession提供的createDataFrame方法应用 Schema 到 RDD 的 RowS（行）.
例如:
importorg.apache.spark.sql.types._// Create an RDDvalpeopleRDD=spark.sparkContext.textFile("examples/src/main/resources/people.txt")// The schema is encoded in a stringvalschemaString="name age"// Generate the schema based on the string of schemavalfields=schemaString.split(" ").map(fieldName=>StructField(fieldName,StringType,nullable=true))valschema=StructType(fields)// Convert records of the RDD (people) to RowsvalrowRDD=peopleRDD.map(_.split(",")).map(attributes=>Row(attributes(0),attributes(1).trim))// Apply the schema to the RDDvalpeopleDF=spark.createDataFrame(rowRDD,schema)// Creates a temporary view using the DataFramepeopleDF.createOrReplaceTempView("people")// SQL can be run over a temporary view created using DataFramesvalresults=spark.sql("SELECT name FROM people")// The results of SQL queries are DataFrames and support all the normal RDD operations// The columns of a row in the result can be accessed by field index or by field nameresults.map(attributes=>"Name: "+attributes(0)).show()// +-------------+// | value|// +-------------+// |Name: Michael|// | Name: Andy|// | Name: Justin|// +-------------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Thebuilt-in DataFrames functionsprovide common aggregations such ascount(),countDistinct(),avg(),max(),min(), etc. While those functions are designed for DataFrames, Spark SQL also has type-safe versions for some of them inScalaandJavato work with strongly typed Datasets. Moreover, users are not limited to the predefined aggregate functions and can create their own.
Untyped User-Defined Aggregate Functions
Users have to extend theUserDefinedAggregateFunctionabstract class to implement a custom untyped aggregate function. For example, a user-defined average can look like:
importorg.apache.spark.sql.expressions.MutableAggregationBufferimportorg.apache.spark.sql.expressions.UserDefinedAggregateFunctionimportorg.apache.spark.sql.types._importorg.apache.spark.sql.Rowimportorg.apache.spark.sql.SparkSessionobjectMyAverageextendsUserDefinedAggregateFunction{// Data types of input arguments of this aggregate functiondefinputSchema:StructType=StructType(StructField("inputColumn",LongType)::Nil)// Data types of values in the aggregation bufferdefbufferSchema:StructType={StructType(StructField("sum",LongType)::StructField("count",LongType)::Nil)}// The data type of the returned valuedefdataType:DataType=DoubleType// Whether this function always returns the same output on the identical inputdefdeterministic:Boolean=true// Initializes the given aggregation buffer. The buffer itself is a `Row` that in addition to// standard methods like retrieving a value at an index (e.g., get(), getBoolean()), provides// the opportunity to update its values. Note that arrays and maps inside the buffer are still// immutable.definitialize(buffer:MutableAggregationBuffer):Unit={buffer(0)=0Lbuffer(1)=0L}// Updates the given aggregation buffer `buffer` with new input data from `input`defupdate(buffer:MutableAggregationBuffer,input:Row):Unit={if(!input.isNullAt(0)){buffer(0)=buffer.getLong(0)+input.getLong(0)buffer(1)=buffer.getLong(1)+1}}// Merges two aggregation buffers and stores the updated buffer values back to `buffer1`defmerge(buffer1:MutableAggregationBuffer,buffer2:Row):Unit={buffer1(0)=buffer1.getLong(0)+buffer2.getLong(0)buffer1(1)=buffer1.getLong(1)+buffer2.getLong(1)}// Calculates the final resultdefevaluate(buffer:Row):Double=buffer.getLong(0).toDouble/buffer.getLong(1)}// Register the function to access itspark.udf.register("myAverage",MyAverage)valdf=spark.read.json("examples/src/main/resources/employees.json")df.createOrReplaceTempView("employees")df.show()// +-------+------+// | name|salary|// +-------+------+// |Michael| 3000|// | Andy| 4500|// | Justin| 3500|// | Berta| 4000|// +-------+------+valresult=spark.sql("SELECT myAverage(salary) as average_salary FROM employees")result.show()// +--------------+// |average_salary|// +--------------+// | 3750.0|// +--------------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/UserDefinedUntypedAggregation.scala" in the Spark repo.
Type-Safe User-Defined Aggregate Functions
User-defined aggregations for strongly typed Datasets revolve around theAggregatorabstract class. For example, a type-safe user-defined average can look like:
importorg.apache.spark.sql.expressions.Aggregatorimportorg.apache.spark.sql.Encoderimportorg.apache.spark.sql.Encodersimportorg.apache.spark.sql.SparkSessioncaseclassEmployee(name:String,salary:Long)caseclassAverage(varsum:Long,varcount:Long)objectMyAverageextendsAggregator[Employee,Average,Double]{// A zero value for this aggregation. Should satisfy the property that any b + zero = bdefzero:Average=Average(0L,0L)// Combine two values to produce a new value. For performance, the function may modify `buffer`// and return it instead of constructing a new objectdefreduce(buffer:Average,employee:Employee):Average={buffer.sum+=employee.salarybuffer.count+=1buffer}// Merge two intermediate valuesdefmerge(b1:Average,b2:Average):Average={b1.sum+=b2.sumb1.count+=b2.countb1}// Transform the output of the reductiondeffinish(reduction:Average):Double=reduction.sum.toDouble/reduction.count// Specifies the Encoder for the intermediate value typedefbufferEncoder:Encoder[Average]=Encoders.product// Specifies the Encoder for the final output value typedefoutputEncoder:Encoder[Double]=Encoders.scalaDouble}valds=spark.read.json("examples/src/main/resources/employees.json").as[Employee]ds.show()// +-------+------+// | name|salary|// +-------+------+// |Michael| 3000|// | Andy| 4500|// | Justin| 3500|// | Berta| 4000|// +-------+------+// Convert the function to a `TypedColumn` and give it a namevalaverageSalary=MyAverage.toColumn.name("average_salary")valresult=ds.select(averageSalary)result.show()// +--------------+// |average_salary|// +--------------+// | 3750.0|// +--------------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/UserDefinedTypedAggregation.scala" in the Spark repo.
Spark SQL 支持通过 DataFrame 接口对各种 data sources （数据源）进行操作. DataFrame 可以使用 relational transformations （关系转换）操作, 也可用于创建 temporary view （临时视图）. 将 DataFrame 注册为 temporary view （临时视图）允许您对其数据运行 SQL 查询. 本节 描述了使用 Spark Data Sources 加载和保存数据的一般方法, 然后涉及可用于 built-in data sources （内置数据源）的 specific options （特定选项）.
Generic Load/Save Functions （通用 加载/保存 功能）
在最简单的形式中, 默认数据源（parquet, 除非另有配置spark.sql.sources.default）将用于所有操作.
valusersDF=spark.read.load("examples/src/main/resources/users.parquet")usersDF.select("name","favorite_color").write.save("namesAndFavColors.parquet")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Manually Specifying Options （手动指定选项）
您还可以 manually specify （手动指定）将与任何你想传递给 data source 的其他选项一起使用的 data source . Data sources 由其 fully qualified name （完全限定名称）（即org.apache.spark.sql.parquet）, 但是对于 built-in sources （内置的源）, 你也可以使用它们的 shortnames （短名称）（json,parquet,jdbc,orc,libsvm,csv,text）.从任何 data source type （数据源类型）加载 DataFrames 可以使用此 syntax （语法）转换为其他类型.
valpeopleDF=spark.read.format("json").load("examples/src/main/resources/people.json")peopleDF.select("name","age").write.format("parquet").save("namesAndAges.parquet")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Run SQL on files directly （直接在文件上运行 SQL）
不使用读取 API 将文件加载到 DataFrame 并进行查询, 也可以直接用 SQL 查询该文件.
valsqlDF=spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Save operations （保存操作）可以选择使用SaveMode, 它指定如何处理现有数据如果存在的话. 重要的是要意识到, 这些 save modes （保存模式）不使用任何 locking （锁定）并且不是 atomic （原子）. 另外, 当执行Overwrite时, 数据将在新数据写出之前被删除.
Scala/JavaAny LanguageMeaning
SaveMode.ErrorIfExists(default)"error"(default)将 DataFrame 保存到 data source （数据源）时, 如果数据已经存在, 则会抛出异常.
SaveMode.Append"append"将 DataFrame 保存到 data source （数据源）时, 如果 data/table 已存在, 则 DataFrame 的内容将被 append （附加）到现有数据中.
SaveMode.Overwrite"overwrite"Overwrite mode （覆盖模式）意味着将 DataFrame 保存到 data source （数据源）时, 如果 data/table 已经存在, 则预期 DataFrame 的内容将 overwritten （覆盖）现有数据.
SaveMode.Ignore"ignore"Ignore mode （忽略模式）意味着当将 DataFrame 保存到 data source （数据源）时, 如果数据已经存在, 则保存操作预期不会保存 DataFrame 的内容, 并且不更改现有数据. 这与 SQL 中的CREATE TABLE IF NOT EXISTS类似.
Saving to Persistent Tables （保存到持久表）
DataFrames也可以使用saveAsTable命令作为 persistent tables （持久表）保存到 Hive metastore 中. 请注意, existing Hive deployment （现有的 Hive 部署）不需要使用此功能. Spark 将为您创建默认的 local Hive metastore （本地 Hive metastore）（使用 Derby ）. 与createOrReplaceTempView命令不同,saveAsTable将 materialize （实现） DataFrame 的内容, 并创建一个指向 Hive metastore 中数据的指针. 即使您的 Spark 程序重新启动, Persistent tables （持久性表）仍然存在, 因为您保持与同一个 metastore 的连接. 可以通过使用表的名称在SparkSession上调用table方法来创建 persistent tabl （持久表）的 DataFrame .
对于 file-based （基于文件）的 data source （数据源）, 例如 text, parquet, json等, 您可以通过path选项指定 custom table path （自定义表路径）, 例如df.write.option("path", "/some/path").saveAsTable("t"). 当表被 dropped （删除）时, custom table path （自定义表路径）将不会被删除, 并且表数据仍然存在. 如果未指定自定义表路径, Spark 将把数据写入 warehouse directory （仓库目录）下的默认表路径. 当表被删除时, 默认的表路径也将被删除.
从 Spark 2.1 开始, persistent datasource tables （持久性数据源表）将 per-partition metadata （每个分区元数据）存储在 Hive metastore 中. 这带来了几个好处:
由于 metastore 只能返回查询的必要 partitions （分区）, 因此不再需要将第一个查询上的所有 partitions discovering 到表中.
Hive DDLs 如ALTER TABLE PARTITION ... SET LOCATION现在可用于使用 Datasource API 创建的表.
请注意, 创建 external datasource tables （外部数据源表）（带有path选项）的表时, 默认情况下不会收集 partition information （分区信息）. 要 sync （同步） metastore 中的分区信息, 可以调用MSCK REPAIR TABLE.
Bucketing, Sorting and Partitioning （分桶, 排序和分区）
对于 file-based data source （基于文件的数据源）, 也可以对 output （输出）进行 bucket 和 sort 或者 partition . Bucketing 和 sorting 仅适用于 persistent tables :
peopleDF.write.bucketBy(42,"name").sortBy("age").saveAsTable("people_bucketed")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
在使用 Dataset API 时, partitioning 可以同时与save和saveAsTable一起使用.
usersDF.write.partitionBy("favorite_color").format("parquet").save("namesPartByColor.parquet")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
可以为 single table （单个表）使用 partitioning 和 bucketing:
peopleDF.write.partitionBy("favorite_color").bucketBy(42,"name").saveAsTable("people_partitioned_bucketed")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
partitionBy创建一个 directory structure （目录结构）, 如Partition Discovery部分所述. 因此, 对 cardinality （基数）较高的 columns 的适用性有限. 相反,bucketBy可以在固定数量的 buckets 中分配数据, 并且可以在 a number of unique values is unbounded （多个唯一值无界时）使用数据.
Parquet是许多其他数据处理系统支持的 columnar format （柱状格式）. Spark SQL 支持读写 Parquet 文件, 可自动保留 schema of the original data （原始数据的模式）. 当编写 Parquet 文件时, 出于兼容性原因, 所有 columns 都将自动转换为可空.
Loading Data Programmatically （以编程的方式加载数据）
使用上面例子中的数据:
// Encoders for most common types are automatically provided by importing spark.implicits._importspark.implicits._valpeopleDF=spark.read.json("examples/src/main/resources/people.json")// DataFrames can be saved as Parquet files, maintaining the schema informationpeopleDF.write.parquet("people.parquet")// Read in the parquet file created above// Parquet files are self-describing so the schema is preserved// The result of loading a Parquet file is also a DataFramevalparquetFileDF=spark.read.parquet("people.parquet")// Parquet files can also be used to create a temporary view and then used in SQL statementsparquetFileDF.createOrReplaceTempView("parquetFile")valnamesDF=spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")namesDF.map(attributes=>"Name: "+attributes(0)).show()// +------------+// | value|// +------------+// |Name: Justin|// +------------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Table partitioning （表分区）是在像 Hive 这样的系统中使用的常见的优化方法. 在 partitioned table （分区表）中, 数据通常存储在不同的目录中, partitioning column values encoded （分区列值编码）在每个 partition directory （分区目录）的路径中. Parquet data source （Parquet 数据源）现在可以自动 discover （发现）和 infer （推断）分区信息. 例如, 我们可以使用以下 directory structure （目录结构）将所有以前使用的 population data （人口数据）存储到 partitioned table （分区表）中, 其中有两个额外的列gender和country作为 partitioning columns （分区列）:
path└── to └── table ├── gender=male │ ├── ... │ │ │ ├── country=US │ │ └── data.parquet │ ├── country=CN │ │ └── data.parquet │ └── ... └── gender=female ├── ... │ ├── country=US │ └── data.parquet ├── country=CN │ └── data.parquet └── ...
通过将path/to/table传递给SparkSession.read.parquet或SparkSession.read.load, Spark SQL 将自动从路径中提取 partitioning information （分区信息）. 现在返回的 DataFrame 的 schema （模式）变成:
root|-- name: string (nullable = true)|-- age: long (nullable = true)|-- gender: string (nullable = true)|-- country: string (nullable = true)
请注意, 会自动 inferred （推断） partitioning columns （分区列）的 data types （数据类型）.目前, 支持 numeric data types （数字数据类型）和 string type （字符串类型）.有些用户可能不想自动推断 partitioning columns （分区列）的数据类型.对于这些用例, automatic type inference （自动类型推断）可以由spark.sql.sources.partitionColumnTypeInference.enabled配置, 默认为true.当禁用 type inference （类型推断）时, string type （字符串类型）将用于 partitioning columns （分区列）.
从 Spark 1.6.0 开始, 默认情况下, partition discovery （分区发现）只能找到给定路径下的 partitions （分区）.对于上述示例, 如果用户将path/to/table/gender=male传递给SparkSession.read.parquet或SparkSession.read.load, 则gender将不被视为 partitioning column （分区列）.如果用户需要指定 partition discovery （分区发现）应该开始的基本路径, 则可以在数据源选项中设置basePath.例如, 当path/to/table/gender=male是数据的路径并且用户将basePath设置为path/to/table/,gender将是一个 partitioning column （分区列）.
像 ProtocolBuffer , Avro 和 Thrift 一样, Parquet 也支持 schema evolution （模式演进）. 用户可以从一个 simple schema （简单的架构）开始, 并根据需要逐渐向 schema 添加更多的 columns （列）. 以这种方式, 用户可能会使用不同但相互兼容的 schemas 的 multiple Parquet files （多个 Parquet 文件）. Parquet data source （Parquet 数据源）现在能够自动检测这种情况并 merge （合并）所有这些文件的 schemas .
由于 schema merging （模式合并）是一个 expensive operation （相对昂贵的操作）, 并且在大多数情况下不是必需的, 所以默认情况下从 1.5.0 开始. 你可以按照如下的方式启用它:
读取 Parquet 文件时, 将 data source option （数据源选项）mergeSchema设置为true（如下面的例子所示）, 或
将 global SQL option （全局 SQL 选项）spark.sql.parquet.mergeSchema设置为true.
// This is used to implicitly convert an RDD to a DataFrame.importspark.implicits._// Create a simple DataFrame, store into a partition directoryvalsquaresDF=spark.sparkContext.makeRDD(1to5).map(i=>(i,i*i)).toDF("value","square")squaresDF.write.parquet("data/test_table/key=1")// Create another DataFrame in a new partition directory,// adding a new column and dropping an existing columnvalcubesDF=spark.sparkContext.makeRDD(6to10).map(i=>(i,i*i*i)).toDF("value","cube")cubesDF.write.parquet("data/test_table/key=2")// Read the partitioned tablevalmergedDF=spark.read.option("mergeSchema","true").parquet("data/test_table")mergedDF.printSchema()// The final schema consists of all 3 columns in the Parquet files together// with the partitioning column appeared in the partition directory paths// root// |-- value: int (nullable = true)// |-- square: int (nullable = true)// |-- cube: int (nullable = true)// |-- key: int (nullable = true)
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Hive metastore Parquet table conversion （Hive metastore Parquet table 转换）
当读取和写入 Hive metastore Parquet 表时, Spark SQL 将尝试使用自己的 Parquet support （Parquet 支持）, 而不是 Hive SerDe 来获得更好的性能. 此 behavior （行为）由spark.sql.hive.convertMetastoreParquet配置控制, 默认情况下 turned on （打开）.
Hive/Parquet Schema Reconciliation
从 table schema processing （表格模式处理）的角度来说, Hive 和 Parquet 之间有两个关键的区别.
Hive 不区分大小写, 而 Parquet 不是
Hive 认为所有 columns （列）都可以为空, 而 Parquet 中的可空性是 significant （重要）的.
由于这个原因, 当将 Hive metastore Parquet 表转换为 Spark SQL Parquet 表时, 我们必须调整 metastore schema 与 Parquet schema. reconciliation 规则是:
在两个 schema 中具有 same name （相同名称）的 Fields （字段）必须具有 same data type （相同的数据类型）, 而不管 nullability （可空性）. reconciled field 应具有 Parquet 的数据类型, 以便 nullability （可空性）得到尊重.
reconciled schema （调和模式）正好包含 Hive metastore schema 中定义的那些字段.
只出现在 Parquet schema 中的任何字段将被 dropped （删除）在 reconciled schema 中.
仅在 Hive metastore schema 中出现的任何字段在 reconciled schema 中作为 nullable field （可空字段）添加.
Spark SQL 缓存 Parquet metadata 以获得更好的性能. 当启用 Hive metastore Parquet table conversion （转换）时, 这些 converted tables （转换表）的 metadata （元数据）也被 cached （缓存）. 如果这些表由 Hive 或其他外部工具更新, 则需要手动刷新以确保 consistent metadata （一致的元数据）.
// spark is an existing SparkSessionspark.catalog.refreshTable("my_table")
可以使用SparkSession上的setConf方法或使用 SQL 运行SET key = value命令来完成 Parquet 的配置.
Property Name （参数名称）Default（默认）Meaning（含义）
spark.sql.parquet.binaryAsStringfalse一些其他 Parquet-producing systems （Parquet 生产系统）, 特别是 Impala, Hive 和旧版本的 Spark SQL , 在 writing out （写出） Parquet schema 时, 不区分 binary data （二进制数据）和 strings （字符串）. 该 flag 告诉 Spark SQL 将 binary data （二进制数据）解释为 string （字符串）以提供与这些系统的兼容性.
spark.sql.parquet.int96AsTimestamptrue一些 Parquet-producing systems , 特别是 Impala 和 Hive , 将 Timestamp 存入INT96 . 该 flag 告诉 Spark SQL 将 INT96 数据解析为 timestamp 以提供与这些系统的兼容性.
spark.sql.parquet.cacheMetadatatrue打开 Parquet schema metadata 的缓存. 可以加快查询静态数据.
spark.sql.parquet.compression.codecsnappy在编写 Parquet 文件时设置 compression codec （压缩编解码器）的使用. 可接受的值包括: uncompressed, snappy, gzip, lzo .
spark.sql.parquet.filterPushdowntrue设置为 true 时启用 Parquet filter push-down optimization .
spark.sql.hive.convertMetastoreParquettrue当设置为 false 时, Spark SQL 将使用 Hive SerDe 作为 parquet tables , 而不是内置的支持.
spark.sql.parquet.mergeSchemafalse当为 true 时, Parquet data source （Parquet 数据源） merges （合并）从所有 data files （数据文件）收集的 schemas , 否则如果没有可用的 summary file , 则从 summary file 或 random data file 中挑选 schema .
spark.sql.optimizer.metadataOnlytrue如果为 true , 则启用使用表的 metadata 的 metadata-only query optimization 来生成 partition columns （分区列）而不是 table scans （表扫描）. 当 scanned （扫描）的所有 columns （列）都是 partition columns （分区列）并且 query （查询）具有满足 distinct semantics （不同语义）的 aggregate operator （聚合运算符）时, 它将适用.
Spark SQL 可以 automatically infer （自动推断）JSON dataset 的 schema, 并将其作为Dataset[Row]加载. 这个 conversion （转换）可以在Dataset[String]上使用SparkSession.read.json()来完成, 或 JSON 文件.
请注意, 以a json file提供的文件不是典型的 JSON 文件. 每行必须包含一个 separate （单独的）, self-contained valid （独立的有效的）JSON 对象. 有关更多信息, 请参阅JSON Lines text format, also called newline-delimited JSON.
对于 regular multi-line JSON file （常规的多行 JSON 文件）, 将multiLine选项设置为true.
// Primitive types (Int, String, etc) and Product types (case classes) encoders are// supported by importing this when creating a Dataset.importspark.implicits._// A JSON dataset is pointed to by path.// The path can be either a single text file or a directory storing text filesvalpath="examples/src/main/resources/people.json"valpeopleDF=spark.read.json(path)// The inferred schema can be visualized using the printSchema() methodpeopleDF.printSchema()// root// |-- age: long (nullable = true)// |-- name: string (nullable = true)// Creates a temporary view using the DataFramepeopleDF.createOrReplaceTempView("people")// SQL statements can be run by using the sql methods provided by sparkvalteenagerNamesDF=spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")teenagerNamesDF.show()// +------+// | name|// +------+// |Justin|// +------+// Alternatively, a DataFrame can be created for a JSON dataset represented by// a Dataset[String] storing one JSON object per stringvalotherPeopleDataset=spark.createDataset("""{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}"""::Nil)valotherPeople=spark.read.json(otherPeopleDataset)otherPeople.show()// +---------------+----+// | address|name|// +---------------+----+// |[Columbus,Ohio]| Yin|// +---------------+----+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Spark SQL 还支持读取和写入存储在Apache Hive中的数据。 但是，由于 Hive 具有大量依赖关系，因此这些依赖关系不包含在默认 Spark 分发中。 如果在类路径中找到 Hive 依赖项，Spark 将自动加载它们。 请注意，这些 Hive 依赖关系也必须存在于所有工作节点上，因为它们将需要访问 Hive 序列化和反序列化库 (SerDes)，以访问存储在 Hive 中的数据。
通过将hive-site.xml,core-site.xml（用于安全配置）和hdfs-site.xml（用于 HDFS 配置）文件放在conf/中来完成配置。
当使用 Hive 时，必须用 Hive 支持实例化SparkSession，包括连接到持续的 Hive 转移，支持 Hive serdes 和 Hive 用户定义的功能。 没有现有 Hive 部署的用户仍然可以启用 Hive 支持。 当hive-site.xml未配置时，上下文会自动在当前目录中创建metastore_db，并创建由spark.sql.warehouse.dir配置的目录，该目录默认为Spark应用程序当前目录中的spark-warehouse目录 开始了 请注意，自从2.0.0以来，hive-site.xml中的hive.metastore.warehouse.dir属性已被弃用。 而是使用spark.sql.warehouse.dir来指定仓库中数据库的默认位置。 您可能需要向启动 Spark 应用程序的用户授予写权限。å
importjava.io.Fileimportorg.apache.spark.sql.Rowimportorg.apache.spark.sql.SparkSessioncaseclassRecord(key:Int,value:String)// warehouseLocation points to the default location for managed databases and tablesvalwarehouseLocation=newFile("spark-warehouse").getAbsolutePathvalspark=SparkSession.builder().appName("Spark Hive Example").config("spark.sql.warehouse.dir",warehouseLocation).enableHiveSupport().getOrCreate()importspark.implicits._importspark.sqlsql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive")sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")// Queries are expressed in HiveQLsql("SELECT * FROM src").show()// +---+-------+// |key| value|// +---+-------+// |238|val_238|// | 86| val_86|// |311|val_311|// ...// Aggregation queries are also supported.sql("SELECT COUNT(*) FROM src").show()// +--------+// |count(1)|// +--------+// | 500 |// +--------+// The results of SQL queries are themselves DataFrames and support all normal functions.valsqlDF=sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key")// The items in DataFrames are of type Row, which allows you to access each column by ordinal.valstringsDS=sqlDF.map{caseRow(key:Int,value:String)=>s"Key:$key, Value:$value"}stringsDS.show()// +--------------------+// | value|// +--------------------+// |Key: 0, Value: val_0|// |Key: 0, Value: val_0|// |Key: 0, Value: val_0|// ...// You can also use DataFrames to create temporary views within a SparkSession.valrecordsDF=spark.createDataFrame((1to100).map(i=>Record(i,s"val_$i")))recordsDF.createOrReplaceTempView("records")// Queries can then join DataFrame data with data stored in Hive.sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()// +---+------+---+------+// |key| value|key| value|// +---+------+---+------+// | 2| val_2| 2| val_2|// | 4| val_4| 4| val_4|// | 5| val_5| 5| val_5|// ...
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/hive/SparkHiveExample.scala" in the Spark repo.
创建 Hive 表时，需要定义如何 从/向 文件系统 read/write 数据，即 “输入格式” 和 “输出格式”。 您还需要定义该表如何将数据反序列化为行，或将行序列化为数据，即 “serde”。 以下选项可用于指定存储格式 (“serde”, “input format”, “output format”)，例如，CREATE TABLE src(id int) USING hive OPTIONS(fileFormat 'parquet')。 默认情况下，我们将以纯文本形式读取表格文件。 请注意，Hive 存储处理程序在创建表时不受支持，您可以使用 Hive 端的存储处理程序创建一个表，并使用 Spark SQL 来读取它。
Property NameMeaning
fileFormatfileFormat是一种存储格式规范的包，包括 "serde"，"input format" 和 "output format"。 目前我们支持6个文件格式：'sequencefile'，'rcfile'，'orc'，'parquet'，'textfile'和'avro'。
inputFormat, outputFormat这两个选项将相应的 "InputFormat" 和 "OutputFormat" 类的名称指定为字符串文字，例如: `org.apache.hadoop.hive.ql.io.orc.OrcInputFormat`。 这两个选项必须成对出现，如果您已经指定了 "fileFormat" 选项，则无法指定它们。
serde此选项指定 serde 类的名称。 当指定 `fileFormat` 选项时，如果给定的 `fileFormat` 已经包含 serde 的信息，那么不要指定这个选项。 目前的 "sequencefile", "textfile" 和 "rcfile" 不包含 serde 信息，你可以使用这3个文件格式的这个选项。
fieldDelim, escapeDelim, collectionDelim, mapkeyDelim, lineDelim这些选项只能与 "textfile" 文件格式一起使用。它们定义如何将分隔的文件读入行。
使用OPTIONS定义的所有其他属性将被视为 Hive serde 属性。
Spark SQL 的 Hive 支持的最重要的部分之一是与 Hive metastore 进行交互，这使得 Spark SQL 能够访问 Hive 表的元数据。 从 Spark 1.4.0 开始，使用 Spark SQL 的单一二进制构建可以使用下面所述的配置来查询不同版本的 Hive 转移。 请注意，独立于用于与转移点通信的 Hive 版本，内部 Spark SQL 将针对 Hive 1.2.1 进行编译，并使用这些类进行内部执行（serdes，UDF，UDAF等）。
以下选项可用于配置用于检索元数据的 Hive 版本：
属性名称默认值含义
spark.sql.hive.metastore.version1.2.1Hive metastore 版本。 可用选项为0.12.0至1.2.1。
spark.sql.hive.metastore.jarsbuiltin当启用-Phive时，使用 Hive 1.2.1，它与 Spark 程序集捆绑在一起。选择此选项时，spark.sql.hive.metastore.version 必须为1.2.1或未定义。 行家 使用从Maven存储库下载的指定版本的Hive jar。 通常不建议在生产部署中使用此配置。 ***** 应用于实例化 HiveMetastoreClient 的 jar 的位置。该属性可以是三个选项之一:
builtin当启用-Phive时，使用 Hive 1.2.1，它与 Spark 程序集捆绑在一起。选择此选项时，spark.sql.hive.metastore.version必须为1.2.1或未定义。
maven使用从 Maven 存储库下载的指定版本的 Hive jar。通常不建议在生产部署中使用此配置。
JVM 的标准格式的 classpath。 该类路径必须包含所有 Hive 及其依赖项，包括正确版本的 Hadoop。这些罐只需要存在于 driver 程序中，但如果您正在运行在 yarn 集群模式，那么您必须确保它们与应用程序一起打包。
spark.sql.hive.metastore.sharedPrefixescom.mysql.jdbc,
org.postgresql,
com.microsoft.sqlserver,
oracle.jdbc使用逗号分隔的类前缀列表，应使用在 Spark SQL 和特定版本的 Hive 之间共享的类加载器来加载。 一个共享类的示例就是用来访问 Hive metastore 的 JDBC driver。 其它需要共享的类，是需要与已经共享的类进行交互的。 例如，log4j 使用的自定义 appender。
spark.sql.hive.metastore.barrierPrefixes(empty)一个逗号分隔的类前缀列表，应该明确地为 Spark SQL 正在通信的 Hive 的每个版本重新加载。 例如，在通常将被共享的前缀中声明的 Hive UDF （即： �org.apache.spark.*）。
Spark SQL 还包括可以使用 JDBC 从其他数据库读取数据的数据源。此功能应优于使用JdbcRDD。 这是因为结果作为 DataFrame 返回，并且可以轻松地在 Spark SQL 中处理或与其他数据源连接。 JDBC 数据源也更容易从 Java 或 Python 使用，因为它不需要用户提供 ClassTag。（请注意，这不同于 Spark SQL JDBC 服务器，允许其他应用程序使用 Spark SQL 运行查询）。
要开始使用，您需要在 Spark 类路径中包含特定数据库的 JDBC driver 程序。 例如，要从 Spark Shell 连接到 postgres，您将运行以下命令:
bin/spark-shell --driver-class-path postgresql-9.4.1207.jar --jars postgresql-9.4.1207.jar
可以使用 Data Sources API 将来自远程数据库的表作为 DataFrame 或 Spark SQL 临时视图进行加载。 用户可以在数据源选项中指定 JDBC 连接属性。用户和密码通常作为登录数据源的连接属性提供。 除了连接属性外，Spark 还支持以下不区分大小写的选项:
�属性名称含义
url要连接的JDBC URL。 源特定的连接属性可以在URL中指定。 例如jdbc：jdbc:postgresql://localhost/test?user=fred&password=secret
dbtable应该读取的 JDBC 表。请注意，可以使用在SQL查询的FROM子句中有效的任何内容。 例如，您可以使用括号中的子查询代替完整表。
driver用于连接到此 URL 的 JDBC driver 程序的类名。
partitionColumn, lowerBound, upperBound如果指定了这些选项，则必须指定这些选项。 另外，必须指定numPartitions. 他们描述如何从多个 worker 并行读取数据时将表给分区。partitionColumn必须是有问题的表中的数字列。 请注意，lowerBound和upperBound仅用于决定分区的大小，而不是用于过滤表中的行。 因此，表中的所有行将被分区并返回。此选项仅适用于读操作。
numPartitions在表读写中可以用于并行度的最大分区数。这也确定并发JDBC连接的最大数量。 如果要写入的分区数超过此限制，则在写入之前通过调用coalesce(numPartitions)将其减少到此限制。
fetchsizeJDBC 抓取的大小，用于确定每次数据往返传递的行数。 这有利于提升 JDBC driver 的性能，它们的默认值较小（例如: Oracle 是 10 行）。 该选项仅适用于读取操作。
batchsizeJDBC 批处理的大小，用于确定每次数据往返传递的行数。 这有利于提升 JDBC driver 的性能。 该选项仅适用于写操作。默认值为1000.
isolationLevel事务隔离级别，适用于当前连接。 它可以是NONE,READ_COMMITTED,READ_UNCOMMITTED,REPEATABLE_READ, 或SERIALIZABLE之一，对应于 JDBC 连接对象定义的标准事务隔离级别，默认为READ_UNCOMMITTED。 此选项仅适用于写操作。请参考java.sql.Connection中的文档。
truncate这是一个与 JDBC 相关的选项。 启用SaveMode.Overwrite时，此选项会导致 Spark 截断现有表，而不是删除并重新创建。 这可以更有效，并且防止表元数据（例如，索引）被移除。 但是，在某些情况下，例如当新数据具有不同的模式时，它将无法工作。 它默认为false。 此选项仅适用于写操作。
createTableOptions这是一个与JDBC相关的选项。 如果指定，此选项允许在创建表时设置特定于数据库的表和分区选项（例如：CREATE TABLE t (name string) ENGINE=InnoDB.）。此选项仅适用于写操作。
createTableColumnTypes使用数据库列数据类型而不是默认值，创建表时。 数据类型信息应以与 CREATE TABLE 列语法相同的格式指定（例如："name CHAR(64), comments VARCHAR(1024)"）。 指定的类型应该是有效的 spark sql 数据类型。此选项仅适用于写操作。
// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods// Loading data from a JDBC sourcevaljdbcDF=spark.read.format("jdbc").option("url","jdbc:postgresql:dbserver").option("dbtable","schema.tablename").option("user","username").option("password","password").load()valconnectionProperties=newProperties()connectionProperties.put("user","username")connectionProperties.put("password","password")valjdbcDF2=spark.read.jdbc("jdbc:postgresql:dbserver","schema.tablename",connectionProperties)// Saving data to a JDBC sourcejdbcDF.write.format("jdbc").option("url","jdbc:postgresql:dbserver").option("dbtable","schema.tablename").option("user","username").option("password","password").save()jdbcDF2.write.jdbc("jdbc:postgresql:dbserver","schema.tablename",connectionProperties)// Specifying create table column data types on writejdbcDF.write.option("createTableColumnTypes","name CHAR(64), comments VARCHAR(1024)").jdbc("jdbc:postgresql:dbserver","schema.tablename",connectionProperties)
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
JDBC driver 程序类必须对客户端会话和所有执行程序上的原始类加载器可见。 这是因为 Java 的 DriverManager 类执行安全检查，导致它忽略原始类加载器不可见的所有 driver 程序，当打开连接时。一个方便的方法是修改所有工作节点上的compute_classpath.sh 以包含您的 driver 程序 JAR。
一些数据库，例如 H2，将所有名称转换为大写。 您需要使用大写字母来引用 Spark SQL 中的这些名称。
对于某些工作负载，可以通过缓存内存中的数据或打开一些实验选项来提高性能。
Spark SQL 可以通过调用spark.catalog.cacheTable("tableName")或dataFrame.cache()来使用内存中的列格式来缓存表。 然后，Spark SQL 将只扫描所需的列，并将自动调整压缩以最小化内存使用量和 GC 压力。 您可以调用spark.catalog.uncacheTable("tableName")从内存中删除该表。
内存缓存的配置可以使用SparkSession上的setConf方法或使用 SQL 运行SET key=value命令来完成。
属性名称默认含义
spark.sql.inMemoryColumnarStorage.compressedtrue当设置为 true 时，Spark SQL 将根据数据的统计信息为每个列自动选择一个压缩编解码器。
spark.sql.inMemoryColumnarStorage.batchSize10000控制批量的柱状缓存的大小。更大的批量大小可以提高内存利用率和压缩率，但是在缓存数据时会冒出 OOM 风险。
以下选项也可用于调整查询执行的性能。这些选项可能会在将来的版本中被废弃，因为更多的优化是自动执行的。
属性名称默认值含义
spark.sql.files.maxPartitionBytes134217728 (128 MB)在读取文件时，将单个分区打包的最大字节数。
spark.sql.files.openCostInBytes4194304 (4 MB)按照字节数来衡量的打开文件的估计费用可以在同一时间进行扫描。 将多个文件放入分区时使用。最好过度估计，那么具有小文件的分区将比具有较大文件的分区（首先计划的）更快。
spark.sql.broadcastTimeout300广播连接中的广播等待时间超时（秒）
spark.sql.autoBroadcastJoinThreshold10485760 (10 MB)配置执行连接时将广播给所有工作节点的表的最大大小（以字节为单位）。 通过将此值设置为-1可以禁用广播。 请注意，目前的统计信息仅支持 Hive Metastore 表，其中已运行命令ANALYZE TABLE COMPUTE STATISTICS noscan。
spark.sql.shuffle.partitions200Configures the number of partitions to use when shuffling data for joins or aggregations.
Spark SQL 也可以充当使用其 JDBC/ODBC 或命令行界面的分布式查询引擎。 在这种模式下，最终用户或应用程序可以直接与 Spark SQL 交互运行 SQL 查询，而不需要编写任何代码。
这里实现的 Thrift JDBC/ODBC 服务器对应于 Hive 1.2 中的HiveServer2。 您可以使用 Spark 或 Hive 1.2.1 附带的直线脚本测试 JDBC 服务器。
要启动 JDBC/ODBC 服务器，请在 Spark 目录中运行以下命令:
./sbin/start-thriftserver.sh
此脚本接受所有bin/spark-submit命令行选项，以及--hiveconf选项来指定 Hive 属性。 您可以运行./sbin/start-thriftserver.sh --help查看所有可用选项的完整列表。 默认情况下，服务器监听 localhost:10000. 您可以通过环境变量覆盖此行为，即:
exportHIVE_SERVER2_THRIFT_PORT=exportHIVE_SERVER2_THRIFT_BIND_HOST=./sbin/start-thriftserver.sh\--master \...
or system properties:
./sbin/start-thriftserver.sh\--hiveconf hive.server2.thrift.port=\--hiveconf hive.server2.thrift.bind.host=\--master ...
现在，您可以使用 beeline 来测试 Thrift JDBC/ODBC 服务器:
./bin/beeline
使用 beeline 方式连接到 JDBC/ODBC 服务器:
beeline> !connect jdbc:hive2://localhost:10000
Beeline 将要求您输入用户名和密码。 在非安全模式下，只需输入机器上的用户名和空白密码即可。 对于安全模式，请按照beeline 文档中的说明进行操作。
配置Hive是通过将hive-site.xml,core-site.xml和hdfs-site.xml文件放在conf/中完成的。
您也可以使用 Hive 附带的 beeline 脚本。
Thrift JDBC 服务器还支持通过 HTTP 传输发送 thrift RPC 消息。 使用以下设置启用 HTTP 模式作为系统属性或在conf/中的hive-site.xml文件中启用:
hive.server2.transport.mode - Set this to value: http
hive.server2.thrift.http.port - HTTP port number to listen on; default is 10001
hive.server2.http.endpoint - HTTP endpoint; default is cliservice
要测试，请使用 beeline 以 http 模式连接到 JDBC/ODBC 服务器:
beeline> !connect jdbc:hive2://:/?hive.server2.transport.mode=http;hive.server2.thrift.http.path=
Spark SQL CLI 是在本地模式下运行 Hive 转移服务并执行从命令行输入的查询的方便工具。 请注意，Spark SQL CLI 不能与 Thrift JDBC 服务器通信。
要启动 Spark SQL CLI，请在 Spark 目录中运行以下命令:
./bin/spark-sql
配置 Hive 是通过将hive-site.xml,core-site.xml和hdfs-site.xml文件放在conf/中完成的。 您可以运行./bin/spark-sql --help获取所有可用选项的完整列表。
Spark 2.1.1 介绍了一个新的配置 key:spark.sql.hive.caseSensitiveInferenceMode. 它的默认设置是NEVER_INFER, 其行为与 2.1.0 保持一致. 但是，Spark 2.2.0 将此设置的默认值更改为 “INFER_AND_SAVE”，以恢复与底层文件 schema（模式）具有大小写混合的列名称的 Hive metastore 表的兼容性。使用INFER_AND_SAVE配置的 value, 在第一次访问 Spark 将对其尚未保存推测 schema（模式）的任何 Hive metastore 表执行 schema inference（模式推断）. 请注意，对于具有数千个 partitions（分区）的表，模式推断可能是非常耗时的操作。如果不兼容大小写混合的列名，您可以安全地将spark.sql.hive.caseSensitiveInferenceMode设置为NEVER_INFER，以避免模式推断的初始开销。请注意，使用新的默认INFER_AND_SAVE设置，模式推理的结果被保存为 metastore key 以供将来使用。因此，初始模式推断仅发生在表的第一次访问。
Datasource tables（数据源表）现在存储了 Hive metastore 中的 partition metadata（分区元数据）. 这意味着诸如ALTER TABLE PARTITION ... SET LOCATION这样的 Hive DDLs 现在使用 Datasource API 可用于创建 tables（表）.
遗留的数据源表可以通过MSCK REPAIR TABLE命令迁移到这种格式。建议迁移遗留表利用 Hive DDL 的支持和提供的计划性能。
要确定表是否已迁移，当在表上发出DESCRIBE FORMATTED命令时请查找PartitionProvider: Catalog属性.
Datasource tables（数据源表）的INSERT OVERWRITE TABLE ... PARTITION ...行为的更改。
在以前的 Spark 版本中，INSERT OVERWRITE覆盖了整个 Datasource table，即使给出一个指定的 partition. 现在只有匹配规范的 partition 被覆盖。
请注意，这仍然与 Hive 表的行为不同，Hive 表仅覆盖与新插入数据重叠的分区。
SparkSession现在是 Spark 新的切入点, 它替代了老的SQLContext和HiveContext。注意 : 为了向下兼容，老的 SQLContext 和 HiveContext 仍然保留。可以从SparkSession获取一个新的catalog接口 — 现有的访问数据库和表的 API，如listTables，createExternalTable，dropTempView，cacheTable都被移到该接口。
Dataset API 和 DataFrame API 进行了统一。在 Scala 中，DataFrame变成了Dataset[Row]类型的一个别名，而 Java API 使用者必须将DataFrame替换成Dataset。Dataset 类既提供了强类型转换操作（如map，filter以及groupByKey）也提供了非强类型转换操作（如select和groupBy）。由于编译期的类型安全不是 Python 和 R 语言的一个特性，Dataset 的概念并不适用于这些语言的 API。相反，DataFrame仍然是最基本的编程抽象, 就类似于这些语言中单节点 data frame 的概念。
Dataset 和 DataFrame API 中 unionAll 已经过时并且由union替代。
Dataset 和 DataFrame API 中 explode 已经过时，作为选择，可以结合 select 或 flatMap 使用functions.explode()。
Dataset 和 DataFrame API 中registerTempTable已经过时并且由createOrReplaceTempView替代。
对 Hive tablesCREATE TABLE ... LOCATION行为的更改.
从 Spark 2.0 开始，CREATE TABLE ... LOCATION与CREATE EXTERNAL TABLE ... LOCATION是相同的，以防止意外丢弃用户提供的 locations（位置）中的现有数据。这意味着，在用户指定位置的 Spark SQL 中创建的 Hive 表始终是 Hive 外部表。删除外部表将不会删除数据。 用户不能指定 Hive managed tables（管理表）的位置. 请注意，这与Hive行为不同。
因此，这些表上的 “DROP TABLE” 语句不会删除数据。
从 Spark 1.6 开始，默认情况下服务器在多 session（会话）模式下运行。这意味着每个 JDBC/ODBC 连接拥有一份自己的 SQL 配置和临时函数注册。缓存表仍在并共享。如果您希望以旧的单会话模式运行 Thrift server，请设置选项spark.sql.hive.thriftServer.singleSession为true。您既可以将此选项添加到spark-defaults.conf，或者通过--conf将它传递给start-thriftserver.sh。
./sbin/start-thriftserver.sh\--conf spark.sql.hive.thriftServer.singleSession=true\...
从 1.6.1 开始，在 sparkR 中 withColumn 方法支持添加一个新列或更换 DataFrame 同名的现有列。
从 Spark 1.6 开始，LongType 强制转换为 TimestampType 期望是秒，而不是微秒。这种更改是为了匹配 Hive 1.2 的行为，以便从 numeric（数值）类型进行更一致的类型转换到 TimestampType。更多详情请参阅SPARK-11724。
使用手动管理的内存优化执行，现在是默认启用的，以及代码生成表达式求值。这些功能既可以通过设置spark.sql.tungsten.enabled为false来禁止使用。
Parquet 的模式合并默认情况下不再启用。它可以通过设置spark.sql.parquet.mergeSchema到true以重新启用。
字符串在 Python 列的 columns（列）现在支持使用点（.）来限定列或访问嵌套值。例如df['table.column.nestedField']。但是，这意味着如果你的列名中包含任何圆点，你现在必须避免使用反引号（如table.column.with.dots.nested）。
在内存中的列存储分区修剪默认是开启的。它可以通过设置spark.sql.inMemoryColumnarStorage.partitionPruning为false来禁用。
无限精度的小数列不再支持，而不是 Spark SQL 最大精度为 38 。当从BigDecimal对象推断模式时，现在使用（38，18）。在 DDL 没有指定精度时，则默认保留Decimal(10, 0)。
时间戳现在存储在 1 微秒的精度，而不是 1 纳秒的。
在 sql 语句中，floating point（浮点数）现在解析为 decimal。HiveQL 解析保持不变。
SQL / DataFrame 函数的规范名称现在是小写（例如 sum vs SUM）。
JSON 数据源不会自动加载由其他应用程序（未通过 Spark SQL 插入到数据集的文件）创建的新文件。对于 JSON 持久表（即表的元数据存储在 Hive Metastore），用户可以使用REFRESH TABLESQL 命令或HiveContext的refreshTable方法，把那些新文件列入到表中。对于代表一个 JSON dataset 的 DataFrame，用户需要重新创建 DataFrame，同时 DataFrame 中将包括新的文件。
PySpark 中 DataFrame 的 withColumn 方法支持添加新的列或替换现有的同名列。
DataFrame data reader/writer interface
基于用户反馈，我们创建了一个新的更流畅的 API，用于读取 (SQLContext.read) 中的数据并写入数据 (DataFrame.write), 并且旧的 API 将过时（例如，SQLContext.parquetFile,SQLContext.jsonFile）.
针对SQLContext.read(Scala,Java,Python) 和DataFrame.write(Scala,Java,Python) 的更多细节，请看 API 文档.
DataFrame.groupBy 保留 grouping columns（分组的列）
根据用户的反馈， 我们更改了DataFrame.groupBy().agg()的默认行为以保留DataFrame结果中的 grouping columns（分组列）. 为了在 1.3 中保持该行为，请设置spark.sql.retainGroupColumns为false.
// In 1.3.x, in order for the grouping column "department" to show up,// it must be included explicitly as part of the agg function call.df.groupBy("department").agg($"department",max("age"),sum("expense"))// In 1.4+, grouping column "department" is included automatically.df.groupBy("department").agg(max("age"),sum("expense"))// Revert to 1.3 behavior (not retaining grouping column) by:sqlContext.setConf("spark.sql.retainGroupColumns","false")
之前 1.4 版本中，DataFrame.withColumn() 只支持添加列。该列将始终在 DateFrame 结果中被加入作为新的列，即使现有的列可能存在相同的名称。从 1.4 版本开始，DataFrame.withColumn() 支持添加与所有现有列的名称不同的列或替换现有的同名列。
请注意，这一变化仅适用于 Scala API，并不适用于 PySpark 和 SparkR。
在 Spark 1.3 中，我们从 Spark SQL 中删除了 “Alpha” 的标签，作为一部分已经清理过的可用的 API 。从 Spark 1.3 版本以上，Spark SQL 将提供在 1.X 系列的其他版本的二进制兼容性。这种兼容性保证不包括被明确标记为不稳定的（即 DeveloperApi 类或 Experimental） API。
升级到 Spark SQL 1.3 版本时，用户会发现最大的变化是，SchemaRDD已更名为DataFrame。这主要是因为 DataFrames 不再从 RDD 直接继承，而是由 RDDS 自己来实现这些功能。DataFrames 仍然可以通过调用.rdd方法转换为 RDDS 。
在 Scala 中，有一个从SchemaRDD到DataFrame类型别名，可以为一些情况提供源代码兼容性。它仍然建议用户更新他们的代码以使用DataFrame来代替。Java 和 Python 用户需要更新他们的代码。
此前 Spark 1.3 有单独的Java兼容类（JavaSQLContext和JavaSchemaRDD），借鉴于 Scala API。在 Spark 1.3 中，Java API 和 Scala API 已经统一。两种语言的用户可以使用SQLContext和DataFrame。一般来说论文类尝试使用两种语言的共有类型（如Array替代了一些特定集合）。在某些情况下不通用的类型情况下，（例如，passing in closures 或 Maps）使用函数重载代替。
此外，该 Java 的特定类型的 API 已被删除。Scala 和 Java 的用户可以使用存在于org.apache.spark.sql.types类来描述编程模式。
许多 Spark 1.3 版本以前的代码示例都以import sqlContext._开始，这提供了从 sqlContext 范围的所有功能。在 Spark 1.3 中，我们移除了从RDDs 到DateFrame再到SQLContext内部对象的隐式转换。用户现在应该写成import sqlContext.implicits._.
此外，隐式转换现在只能使用方法toDF来增加由Product（即 case classes or tuples）构成的RDD，而不是自动应用。
当使用 DSL 内部的函数时（现在使用DataFrameAPI 来替换）, 用户习惯导入org.apache.spark.sql.catalyst.dsl. 相反，应该使用公共的 dataframe 函数 API:import org.apache.spark.sql.functions._.
针对 DataType 删除在 org.apache.spark.sql 包中的一些类型别名（仅限于 Scala）
Spark 1.3 移除存在于基本 SQL 包的DataType类型别名。开发人员应改为导入类org.apache.spark.sql.types。
UDF 注册迁移到sqlContext.udf中 (Java & Scala)
用于注册 UDF 的函数，不管是 DataFrame DSL 还是 SQL 中用到的，都被迁移到SQLContext中的 udf 对象中。
sqlContext.udf.register("strLen",(s:String)=>s.length())
Python UDF 注册保持不变。
Python DataTypes 不再是 Singletons（单例的）
在 Python 中使用 DataTypes 时，你需要先构造它们（如：StringType()），而不是引用一个单例对象。
Spark SQL 在设计时就考虑到了和 Hive metastore，SerDes 以及 UDF 之间的兼容性。目前 Hive SerDes 和 UDF 都是基于 Hive 1.2.1 版本，并且Spark SQL 可以连接到不同版本的Hive metastore（从 0.12.0 到 1.2.1，可以参考与不同版本的 Hive Metastore 交互）
Spark SQL Thrift JDBC server 采用了开箱即用的设计以兼容已有的 Hive 安装版本。你不需要修改现有的 Hive Metastore , 或者改变数据的位置和表的分区。
Spark SQL 支持绝大部分的 Hive 功能，如:
Hive query（查询）语句, 包括:
SELECT
GROUP BY
ORDER BY
CLUSTER BY
SORT BY
所有 Hive 操作, 包括:
关系运算符 (=,⇔,==,<>,<,>,>=,<=, 等等)
算术运算符 (+,-,*,/,%, 等等)
逻辑运算符 (AND,&&,OR,||, 等等)
复杂类型的构造
数学函数 (sign,ln,cos, 等等)
String 函数 (instr,length,printf, 等等)
用户定义函数 (UDF)
用户定义聚合函数 (UDAF)
用户定义 serialization formats (SerDes)
窗口函数
Joins
JOIN
{LEFT|RIGHT|FULL} OUTER JOIN
LEFT SEMI JOIN
CROSS JOIN
Unions
Sub-queries（子查询）
SELECT col FROM ( SELECT a + b AS col from t1) t2
Sampling
Explain
Partitioned tables including dynamic partition insertion
View
所有的 Hive DDL 函数, 包括:
CREATE TABLE
CREATE TABLE AS SELECT
ALTER TABLE
大部分的 Hive Data types（数据类型）, 包括:
TINYINT
SMALLINT
INT
BIGINT
BOOLEAN
FLOAT
DOUBLE
STRING
BINARY
TIMESTAMP
DATE
ARRAY<>
MAP<>
STRUCT<>
以下是目前还不支持的 Hive 函数列表。在 Hive 部署中这些功能大部分都用不到。
主要的 Hive 功能
Tables 使用 buckets 的 Tables: bucket 是 Hive table partition 中的 hash partitioning. Spark SQL 还不支持 buckets.
Esoteric Hive 功能
UNION类型
Unique join
Column 统计信息的收集: Spark SQL does not piggyback scans to collect column statistics at the moment and only supports populating the sizeInBytes field of the hive metastore.
Hive Input/Output Formats
File format for CLI: For results showing back to the CLI, Spark SQL only supports TextOutputFormat.
Hadoop archive
Hive 优化
有少数 Hive 优化还没有包含在 Spark 中。其中一些（比如 indexes 索引）由于 Spark SQL 的这种内存计算模型而显得不那么重要。另外一些在 Spark SQL 未来的版本中会持续跟踪。
Block 级别的 bitmap indexes 和虚拟 columns (用于构建 indexes)
自动为 join 和 groupBy 计算 reducer 个数 : 目前在 Spark SQL 中, 你需要使用 “SET spark.sql.shuffle.partitions=[num_tasks];” 来控制 post-shuffle 的并行度.
仅 Meta-data 的 query: 对于只使用 metadata 就能回答的查询，Spark SQL 仍然会启动计算结果的任务.
Skew data flag: Spark SQL 不遵循 Hive 中 skew 数据的标记.
STREAMTABLEhint in join: Spark SQL 不遵循STREAMTABLEhint.
对于查询结果合并多个小文件: 如果输出的结果包括多个小文件, Hive 可以可选的合并小文件到一些大文件中去，以避免溢出 HDFS metadata. Spark SQL 还不支持这样.
Spark SQL 和 DataFrames 支持下面的数据类型:
Numeric types
ByteType: Represents 1-byte signed integer numbers. The range of numbers is from-128to127.
ShortType: Represents 2-byte signed integer numbers. The range of numbers is from-32768to32767.
IntegerType: Represents 4-byte signed integer numbers. The range of numbers is from-2147483648to2147483647.
LongType: Represents 8-byte signed integer numbers. The range of numbers is from-9223372036854775808to9223372036854775807.
FloatType: Represents 4-byte single-precision floating point numbers.
DoubleType: Represents 8-byte double-precision floating point numbers.
DecimalType: Represents arbitrary-precision signed decimal numbers. Backed internally byjava.math.BigDecimal. ABigDecimalconsists of an arbitrary precision integer unscaled value and a 32-bit integer scale.
String type
StringType: Represents character string values.
Binary type
BinaryType: Represents byte sequence values.
Boolean type
BooleanType: Represents boolean values.
Datetime type
TimestampType: Represents values comprising values of fields year, month, day, hour, minute, and second.
DateType: Represents values comprising values of fields year, month, day.
Complex types
ArrayType(elementType, containsNull): Represents values comprising a sequence of elements with the type ofelementType.containsNullis used to indicate if elements in aArrayTypevalue can havenullvalues.
MapType(keyType, valueType, valueContainsNull): Represents values comprising a set of key-value pairs. The data type of keys are described bykeyTypeand the data type of values are described byvalueType. For aMapTypevalue, keys are not allowed to havenullvalues.valueContainsNullis used to indicate if values of aMapTypevalue can havenullvalues.
StructType(fields): Represents values with the structure described by a sequence ofStructFields (fields).
StructField(name, dataType, nullable): Represents a field in aStructType. The name of a field is indicated byname. The data type of a field is indicated bydataType.nullableis used to indicate if values of this fields can havenullvalues.
Spark SQL 的所有数据类型都在包org.apache.spark.sql.types中. 你可以用下示例示例来访问它们.
importorg.apache.spark.sql.types._
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Data type（数据类型）Scala 中的 Value 类型访问或创建数据类型的 API
ByteTypeByteByteType
ShortTypeShortShortType
IntegerTypeIntIntegerType
LongTypeLongLongType
FloatTypeFloatFloatType
DoubleTypeDoubleDoubleType
DecimalTypejava.math.BigDecimalDecimalType
StringTypeStringStringType
BinaryTypeArray[Byte]BinaryType
BooleanTypeBooleanBooleanType
TimestampTypejava.sql.TimestampTimestampType
DateTypejava.sql.DateDateType
ArrayTypescala.collection.SeqArrayType(elementType, [containsNull])
Note（注意）:containsNull的默认值是true.
MapTypescala.collection.MapMapType(keyType,valueType, [valueContainsNull])
Note（注意）:valueContainsNull的默认值是true.
StructTypeorg.apache.spark.sql.RowStructType(fields)
Note（注意）:fields是 StructFields 的 Seq. 所有, 两个 fields 拥有相同的名称是不被允许的.
StructField该 field（字段）数据类型的 Scala 中的 value 类型 (例如, 数据类型为 IntegerType 的 StructField 是 Int)StructField(name,dataType, [nullable])
Note:nullable的默认值是true.
当处理一些不符合标准浮点数语义的float或double类型时，对于 Not-a-Number(NaN) 需要做一些特殊处理. 具体如下:
NaN = NaN 返回 true.
在 aggregations（聚合）操作中，所有的 NaN values 将被分到同一个组中.
在 join key 中 NaN 可以当做一个普通的值.
NaN 值在升序排序中排到最后，比任何其他数值都大.
Spark SQL, DataFrames and Datasets Guide
无类型的Dataset操作 (aka DataFrame 操作)
Running SQL Queries Programmatically
Untyped User-Defined Aggregate Functions
Type-Safe User-Defined Aggregate Functions
Generic Load/Save Functions （通用 加载/保存 功能）
Manually Specifying Options （手动指定选项）
Run SQL on files directly （直接在文件上运行 SQL）
Saving to Persistent Tables （保存到持久表）
Bucketing, Sorting and Partitioning （分桶, 排序和分区）
Loading Data Programmatically （以编程的方式加载数据）
Hive metastore Parquet table conversion （Hive metastore Parquet table 转换）
Hive/Parquet Schema Reconciliation
DataFrame data reader/writer interface
DataFrame.groupBy 保留 grouping columns（分组的列）
针对 DataType 删除在 org.apache.spark.sql 包中的一些类型别名（仅限于 Scala）
UDF 注册迁移到sqlContext.udf中 (Java & Scala)
Python DataTypes 不再是 Singletons（单例的）
Spark SQL 是 Spark 处理结构化数据的一个模块.与基础的 Spark RDD API 不同, Spark SQL 提供了查询结构化数据及计算结果等信息的接口.在内部, Spark SQL 使用这个额外的信息去执行额外的优化.有几种方式可以跟 Spark SQL 进行交互, 包括 SQL 和 Dataset API.当使用相同执行引擎进行计算时, 无论使用哪种 API / 语言都可以快速的计算.这种统一意味着开发人员能够在基于提供最自然的方式来表达一个给定的 transformation API 之间实现轻松的来回切换不同的 .
该页面所有例子使用的示例数据都包含在 Spark 的发布中, 并且可以使用spark-shell,pysparkshell, 或者sparkRshell来运行.
Spark SQL 的功能之一是执行 SQL 查询.Spark SQL 也能够被用于从已存在的 Hive 环境中读取数据.更多关于如何配置这个特性的信息, 请参考Hive 表这部分. 当以另外的编程语言运行SQL 时, 查询结果将以Dataset/DataFrame的形式返回.您也可以使用命令行或者通过JDBC/ODBC与 SQL 接口交互.
一个 Dataset 是一个分布式的数据集合 Dataset 是在 Spark 1.6 中被添加的新接口, 它提供了 RDD 的优点（强类型化, 能够使用强大的 lambda 函数）与Spark SQL执行引擎的优点.一个 Dataset 可以从 JVM 对象来构造并且使用转换功能（map, flatMap, filter, 等等）. Dataset API 在Scala和Java是可用的.Python 不支持 Dataset API.但是由于 Python 的动态特性, 许多 Dataset API 的优点已经可用了 (也就是说, 你可能通过 name 天生的row.columnName属性访问一行中的字段).这种情况和 R 相似.
一个 DataFrame 是一个Dataset组成的指定列.它的概念与一个在关系型数据库或者在 R/Python 中的表是相等的, 但是有很多优化. DataFrames 可以从大量的sources中构造出来, 比如: 结构化的文本文件, Hive中的表, 外部数据库, 或者已经存在的 RDDs. DataFrame API 可以在 Scala, Java,Python, 和R中实现. 在 Scala 和 Java中, 一个 DataFrame 所代表的是一个多个Row（行）的的 Dataset（数据集合）. 在the Scala API中,DataFrame仅仅是一个Dataset[Row]类型的别名. 然而, 在Java API中, 用户需要去使用Dataset去代表一个DataFrame.
在此文档中, 我们将常常会引用 Scala/Java Datasets 的Rows 作为 DataFrames.
Spark SQL中所有功能的入口点是SparkSession类. 要创建一个SparkSession, 仅使用SparkSession.builder()就可以了:
importorg.apache.spark.sql.SparkSessionvalspark=SparkSession.builder().appName("Spark SQL basic example").config("spark.some.config.option","some-value").getOrCreate()// For implicit conversions like converting RDDs to DataFramesimportspark.implicits._
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Spark 2.0 中的SparkSession为 Hive 特性提供了内嵌的支持, 包括使用 HiveQL 编写查询的能力, 访问 Hive UDF,以及从 Hive 表中读取数据的能力.为了使用这些特性, 你不需要去有一个已存在的 Hive 设置.
在一个SparkSession中, 应用程序可以从一个已经存在的RDD, 从hive表, 或者从Spark数据源中创建一个DataFrames.
举个例子, 下面就是基于一个JSON文件创建一个DataFrame:
valdf=spark.read.json("examples/src/main/resources/people.json")// Displays the content of the DataFrame to stdoutdf.show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
无类型的Dataset操作 (aka DataFrame 操作)
DataFrames 提供了一个特定的语法用在Scala,Java,PythonandR中机构化数据的操作.
正如上面提到的一样, Spark 2.0中, DataFrames在Scala 和 Java API中, 仅仅是多个Rows的Dataset. 这些操作也参考了与强类型的Scala/Java Datasets中的”类型转换” 对应的”无类型转换” .
这里包括一些使用 Dataset 进行结构化数据处理的示例 :
// This import is needed to use the $-notationimportspark.implicits._// Print the schema in a tree formatdf.printSchema()// root// |-- age: long (nullable = true)// |-- name: string (nullable = true)// Select only the "name" columndf.select("name").show()// +-------+// | name|// +-------+// |Michael|// | Andy|// | Justin|// +-------+// Select everybody, but increment the age by 1df.select($"name",$"age"+1).show()// +-------+---------+// | name|(age + 1)|// +-------+---------+// |Michael| null|// | Andy| 31|// | Justin| 20|// +-------+---------+// Select people older than 21df.filter($"age">21).show()// +---+----+// |age|name|// +---+----+// | 30|Andy|// +---+----+// Count people by agedf.groupBy("age").count().show()// +----+-----+// | age|count|// +----+-----+// | 19| 1|// |null| 1|// | 30| 1|// +----+-----+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
能够在 DataFrame 上被执行的操作类型的完整列表请参考API 文档.
除了简单的列引用和表达式之外, DataFrame 也有丰富的函数库, 包括 string 操作, date 算术, 常见的 math 操作以及更多.可用的完整列表请参考DataFrame 函数指南.
Running SQL Queries Programmatically
SparkSession的sql函数可以让应用程序以编程的方式运行 SQL 查询, 并将结果作为一个DataFrame返回.
// Register the DataFrame as a SQL temporary viewdf.createOrReplaceTempView("people")valsqlDF=spark.sql("SELECT * FROM people")sqlDF.show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Spark SQL中的临时视图是session级别的, 也就是会随着session的消失而消失. 如果你想让一个临时视图在所有session中相互传递并且可用, 直到Spark 应用退出, 你可以建立一个全局的临时视图.全局的临时视图存在于系统数据库global_temp中, 我们必须加上库名去引用它, 比如.SELECT * FROM global_temp.view1.
// Register the DataFrame as a global temporary viewdf.createGlobalTempView("people")// Global temporary view is tied to a system preserved database `global_temp`spark.sql("SELECT * FROM global_temp.people").show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+// Global temporary view is cross-sessionspark.newSession().sql("SELECT * FROM global_temp.people").show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Dataset 与 RDD 相似, 然而, 并不是使用 Java 序列化或者 Kryo编码器来序列化用于处理或者通过网络进行传输的对象. 虽然编码器和标准的序列化都负责将一个对象序列化成字节, 编码器是动态生成的代码, 并且使用了一种允许 Spark 去执行许多像 filtering, sorting 以及 hashing 这样的操作, 不需要将字节反序列化成对象的格式.
// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,// you can use custom classes that implement the Product interfacecaseclassPerson(name:String,age:Long)// Encoders are created for case classesvalcaseClassDS=Seq(Person("Andy",32)).toDS()caseClassDS.show()// +----+---+// |name|age|// +----+---+// |Andy| 32|// +----+---+// Encoders for most common types are automatically provided by importing spark.implicits._valprimitiveDS=Seq(1,2,3).toDS()primitiveDS.map(_+1).collect()// Returns: Array(2, 3, 4)// DataFrames can be converted to a Dataset by providing a class. Mapping will be done by namevalpath="examples/src/main/resources/people.json"valpeopleDS=spark.read.json(path).as[Person]peopleDS.show()// +----+-------+// | age| name|// +----+-------+// |null|Michael|// | 30| Andy|// | 19| Justin|// +----+-------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Spark SQL 支持两种不同的方法用于转换已存在的 RDD 成为 Dataset.第一种方法是使用反射去推断一个包含指定的对象类型的 RDD 的 Schema.在你的 Spark 应用程序中当你已知 Schema 时这个基于方法的反射可以让你的代码更简洁.
第二种用于创建 Dataset 的方法是通过一个允许你构造一个 Schema 然后把它应用到一个已存在的 RDD 的编程接口.然而这种方法更繁琐, 当列和它们的类型知道运行时都是未知时它允许你去构造 Dataset.
Spark SQL 的 Scala 接口支持自动转换一个包含 case classes 的 RDD 为 DataFrame.Case class 定义了表的 Schema.Case class 的参数名使用反射读取并且成为了列名.Case class 也可以是嵌套的或者包含像Seq或者Array这样的复杂类型.这个 RDD 能够被隐式转换成一个 DataFrame 然后被注册为一个表.表可以用于后续的 SQL 语句.
// For implicit conversions from RDDs to DataFramesimportspark.implicits._// Create an RDD of Person objects from a text file, convert it to a DataframevalpeopleDF=spark.sparkContext.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(attributes=>Person(attributes(0),attributes(1).trim.toInt)).toDF()// Register the DataFrame as a temporary viewpeopleDF.createOrReplaceTempView("people")// SQL statements can be run by using the sql methods provided by SparkvalteenagersDF=spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")// The columns of a row in the result can be accessed by field indexteenagersDF.map(teenager=>"Name: "+teenager(0)).show()// +------------+// | value|// +------------+// |Name: Justin|// +------------+// or by field nameteenagersDF.map(teenager=>"Name: "+teenager.getAs[String]("name")).show()// +------------+// | value|// +------------+// |Name: Justin|// +------------+// No pre-defined encoders for Dataset[Map[K,V]], define explicitlyimplicitvalmapEncoder=org.apache.spark.sql.Encoders.kryo[Map[String,Any]]// Primitive types and case classes can be also defined as// implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder()// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]teenagersDF.map(teenager=>teenager.getValuesMap[Any](List("name","age"))).collect()// Array(Map("name" -> "Justin", "age" -> 19))
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
当 case class 不能够在执行之前被定义（例如, records 记录的结构在一个 string 字符串中被编码了, 或者一个 text 文本 dataset 将被解析并且不同的用户投影的字段是不一样的）.一个DataFrame可以使用下面的三步以编程的方式来创建.
从原始的 RDD 创建 RDD 的Row（行）;
Step 1 被创建后, 创建 Schema 表示一个StructType匹配 RDD 中的Row（行）的结构.
通过SparkSession提供的createDataFrame方法应用 Schema 到 RDD 的 RowS（行）.
例如:
importorg.apache.spark.sql.types._// Create an RDDvalpeopleRDD=spark.sparkContext.textFile("examples/src/main/resources/people.txt")// The schema is encoded in a stringvalschemaString="name age"// Generate the schema based on the string of schemavalfields=schemaString.split(" ").map(fieldName=>StructField(fieldName,StringType,nullable=true))valschema=StructType(fields)// Convert records of the RDD (people) to RowsvalrowRDD=peopleRDD.map(_.split(",")).map(attributes=>Row(attributes(0),attributes(1).trim))// Apply the schema to the RDDvalpeopleDF=spark.createDataFrame(rowRDD,schema)// Creates a temporary view using the DataFramepeopleDF.createOrReplaceTempView("people")// SQL can be run over a temporary view created using DataFramesvalresults=spark.sql("SELECT name FROM people")// The results of SQL queries are DataFrames and support all the normal RDD operations// The columns of a row in the result can be accessed by field index or by field nameresults.map(attributes=>"Name: "+attributes(0)).show()// +-------------+// | value|// +-------------+// |Name: Michael|// | Name: Andy|// | Name: Justin|// +-------------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Thebuilt-in DataFrames functionsprovide common aggregations such ascount(),countDistinct(),avg(),max(),min(), etc. While those functions are designed for DataFrames, Spark SQL also has type-safe versions for some of them inScalaandJavato work with strongly typed Datasets. Moreover, users are not limited to the predefined aggregate functions and can create their own.
Untyped User-Defined Aggregate Functions
Users have to extend theUserDefinedAggregateFunctionabstract class to implement a custom untyped aggregate function. For example, a user-defined average can look like:
importorg.apache.spark.sql.expressions.MutableAggregationBufferimportorg.apache.spark.sql.expressions.UserDefinedAggregateFunctionimportorg.apache.spark.sql.types._importorg.apache.spark.sql.Rowimportorg.apache.spark.sql.SparkSessionobjectMyAverageextendsUserDefinedAggregateFunction{// Data types of input arguments of this aggregate functiondefinputSchema:StructType=StructType(StructField("inputColumn",LongType)::Nil)// Data types of values in the aggregation bufferdefbufferSchema:StructType={StructType(StructField("sum",LongType)::StructField("count",LongType)::Nil)}// The data type of the returned valuedefdataType:DataType=DoubleType// Whether this function always returns the same output on the identical inputdefdeterministic:Boolean=true// Initializes the given aggregation buffer. The buffer itself is a `Row` that in addition to// standard methods like retrieving a value at an index (e.g., get(), getBoolean()), provides// the opportunity to update its values. Note that arrays and maps inside the buffer are still// immutable.definitialize(buffer:MutableAggregationBuffer):Unit={buffer(0)=0Lbuffer(1)=0L}// Updates the given aggregation buffer `buffer` with new input data from `input`defupdate(buffer:MutableAggregationBuffer,input:Row):Unit={if(!input.isNullAt(0)){buffer(0)=buffer.getLong(0)+input.getLong(0)buffer(1)=buffer.getLong(1)+1}}// Merges two aggregation buffers and stores the updated buffer values back to `buffer1`defmerge(buffer1:MutableAggregationBuffer,buffer2:Row):Unit={buffer1(0)=buffer1.getLong(0)+buffer2.getLong(0)buffer1(1)=buffer1.getLong(1)+buffer2.getLong(1)}// Calculates the final resultdefevaluate(buffer:Row):Double=buffer.getLong(0).toDouble/buffer.getLong(1)}// Register the function to access itspark.udf.register("myAverage",MyAverage)valdf=spark.read.json("examples/src/main/resources/employees.json")df.createOrReplaceTempView("employees")df.show()// +-------+------+// | name|salary|// +-------+------+// |Michael| 3000|// | Andy| 4500|// | Justin| 3500|// | Berta| 4000|// +-------+------+valresult=spark.sql("SELECT myAverage(salary) as average_salary FROM employees")result.show()// +--------------+// |average_salary|// +--------------+// | 3750.0|// +--------------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/UserDefinedUntypedAggregation.scala" in the Spark repo.
Type-Safe User-Defined Aggregate Functions
User-defined aggregations for strongly typed Datasets revolve around theAggregatorabstract class. For example, a type-safe user-defined average can look like:
importorg.apache.spark.sql.expressions.Aggregatorimportorg.apache.spark.sql.Encoderimportorg.apache.spark.sql.Encodersimportorg.apache.spark.sql.SparkSessioncaseclassEmployee(name:String,salary:Long)caseclassAverage(varsum:Long,varcount:Long)objectMyAverageextendsAggregator[Employee,Average,Double]{// A zero value for this aggregation. Should satisfy the property that any b + zero = bdefzero:Average=Average(0L,0L)// Combine two values to produce a new value. For performance, the function may modify `buffer`// and return it instead of constructing a new objectdefreduce(buffer:Average,employee:Employee):Average={buffer.sum+=employee.salarybuffer.count+=1buffer}// Merge two intermediate valuesdefmerge(b1:Average,b2:Average):Average={b1.sum+=b2.sumb1.count+=b2.countb1}// Transform the output of the reductiondeffinish(reduction:Average):Double=reduction.sum.toDouble/reduction.count// Specifies the Encoder for the intermediate value typedefbufferEncoder:Encoder[Average]=Encoders.product// Specifies the Encoder for the final output value typedefoutputEncoder:Encoder[Double]=Encoders.scalaDouble}valds=spark.read.json("examples/src/main/resources/employees.json").as[Employee]ds.show()// +-------+------+// | name|salary|// +-------+------+// |Michael| 3000|// | Andy| 4500|// | Justin| 3500|// | Berta| 4000|// +-------+------+// Convert the function to a `TypedColumn` and give it a namevalaverageSalary=MyAverage.toColumn.name("average_salary")valresult=ds.select(averageSalary)result.show()// +--------------+// |average_salary|// +--------------+// | 3750.0|// +--------------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/UserDefinedTypedAggregation.scala" in the Spark repo.
Spark SQL 支持通过 DataFrame 接口对各种 data sources （数据源）进行操作. DataFrame 可以使用 relational transformations （关系转换）操作, 也可用于创建 temporary view （临时视图）. 将 DataFrame 注册为 temporary view （临时视图）允许您对其数据运行 SQL 查询. 本节 描述了使用 Spark Data Sources 加载和保存数据的一般方法, 然后涉及可用于 built-in data sources （内置数据源）的 specific options （特定选项）.
Generic Load/Save Functions （通用 加载/保存 功能）
在最简单的形式中, 默认数据源（parquet, 除非另有配置spark.sql.sources.default）将用于所有操作.
valusersDF=spark.read.load("examples/src/main/resources/users.parquet")usersDF.select("name","favorite_color").write.save("namesAndFavColors.parquet")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Manually Specifying Options （手动指定选项）
您还可以 manually specify （手动指定）将与任何你想传递给 data source 的其他选项一起使用的 data source . Data sources 由其 fully qualified name （完全限定名称）（即org.apache.spark.sql.parquet）, 但是对于 built-in sources （内置的源）, 你也可以使用它们的 shortnames （短名称）（json,parquet,jdbc,orc,libsvm,csv,text）.从任何 data source type （数据源类型）加载 DataFrames 可以使用此 syntax （语法）转换为其他类型.
valpeopleDF=spark.read.format("json").load("examples/src/main/resources/people.json")peopleDF.select("name","age").write.format("parquet").save("namesAndAges.parquet")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Run SQL on files directly （直接在文件上运行 SQL）
不使用读取 API 将文件加载到 DataFrame 并进行查询, 也可以直接用 SQL 查询该文件.
valsqlDF=spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Save operations （保存操作）可以选择使用SaveMode, 它指定如何处理现有数据如果存在的话. 重要的是要意识到, 这些 save modes （保存模式）不使用任何 locking （锁定）并且不是 atomic （原子）. 另外, 当执行Overwrite时, 数据将在新数据写出之前被删除.
Scala/JavaAny LanguageMeaning
SaveMode.ErrorIfExists(default)"error"(default)将 DataFrame 保存到 data source （数据源）时, 如果数据已经存在, 则会抛出异常.
SaveMode.Append"append"将 DataFrame 保存到 data source （数据源）时, 如果 data/table 已存在, 则 DataFrame 的内容将被 append （附加）到现有数据中.
SaveMode.Overwrite"overwrite"Overwrite mode （覆盖模式）意味着将 DataFrame 保存到 data source （数据源）时, 如果 data/table 已经存在, 则预期 DataFrame 的内容将 overwritten （覆盖）现有数据.
SaveMode.Ignore"ignore"Ignore mode （忽略模式）意味着当将 DataFrame 保存到 data source （数据源）时, 如果数据已经存在, 则保存操作预期不会保存 DataFrame 的内容, 并且不更改现有数据. 这与 SQL 中的CREATE TABLE IF NOT EXISTS类似.
Saving to Persistent Tables （保存到持久表）
DataFrames也可以使用saveAsTable命令作为 persistent tables （持久表）保存到 Hive metastore 中. 请注意, existing Hive deployment （现有的 Hive 部署）不需要使用此功能. Spark 将为您创建默认的 local Hive metastore （本地 Hive metastore）（使用 Derby ）. 与createOrReplaceTempView命令不同,saveAsTable将 materialize （实现） DataFrame 的内容, 并创建一个指向 Hive metastore 中数据的指针. 即使您的 Spark 程序重新启动, Persistent tables （持久性表）仍然存在, 因为您保持与同一个 metastore 的连接. 可以通过使用表的名称在SparkSession上调用table方法来创建 persistent tabl （持久表）的 DataFrame .
对于 file-based （基于文件）的 data source （数据源）, 例如 text, parquet, json等, 您可以通过path选项指定 custom table path （自定义表路径）, 例如df.write.option("path", "/some/path").saveAsTable("t"). 当表被 dropped （删除）时, custom table path （自定义表路径）将不会被删除, 并且表数据仍然存在. 如果未指定自定义表路径, Spark 将把数据写入 warehouse directory （仓库目录）下的默认表路径. 当表被删除时, 默认的表路径也将被删除.
从 Spark 2.1 开始, persistent datasource tables （持久性数据源表）将 per-partition metadata （每个分区元数据）存储在 Hive metastore 中. 这带来了几个好处:
由于 metastore 只能返回查询的必要 partitions （分区）, 因此不再需要将第一个查询上的所有 partitions discovering 到表中.
Hive DDLs 如ALTER TABLE PARTITION ... SET LOCATION现在可用于使用 Datasource API 创建的表.
请注意, 创建 external datasource tables （外部数据源表）（带有path选项）的表时, 默认情况下不会收集 partition information （分区信息）. 要 sync （同步） metastore 中的分区信息, 可以调用MSCK REPAIR TABLE.
Bucketing, Sorting and Partitioning （分桶, 排序和分区）
对于 file-based data source （基于文件的数据源）, 也可以对 output （输出）进行 bucket 和 sort 或者 partition . Bucketing 和 sorting 仅适用于 persistent tables :
peopleDF.write.bucketBy(42,"name").sortBy("age").saveAsTable("people_bucketed")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
在使用 Dataset API 时, partitioning 可以同时与save和saveAsTable一起使用.
usersDF.write.partitionBy("favorite_color").format("parquet").save("namesPartByColor.parquet")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
可以为 single table （单个表）使用 partitioning 和 bucketing:
peopleDF.write.partitionBy("favorite_color").bucketBy(42,"name").saveAsTable("people_partitioned_bucketed")
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
partitionBy创建一个 directory structure （目录结构）, 如Partition Discovery部分所述. 因此, 对 cardinality （基数）较高的 columns 的适用性有限. 相反,bucketBy可以在固定数量的 buckets 中分配数据, 并且可以在 a number of unique values is unbounded （多个唯一值无界时）使用数据.
Parquet是许多其他数据处理系统支持的 columnar format （柱状格式）. Spark SQL 支持读写 Parquet 文件, 可自动保留 schema of the original data （原始数据的模式）. 当编写 Parquet 文件时, 出于兼容性原因, 所有 columns 都将自动转换为可空.
Loading Data Programmatically （以编程的方式加载数据）
使用上面例子中的数据:
// Encoders for most common types are automatically provided by importing spark.implicits._importspark.implicits._valpeopleDF=spark.read.json("examples/src/main/resources/people.json")// DataFrames can be saved as Parquet files, maintaining the schema informationpeopleDF.write.parquet("people.parquet")// Read in the parquet file created above// Parquet files are self-describing so the schema is preserved// The result of loading a Parquet file is also a DataFramevalparquetFileDF=spark.read.parquet("people.parquet")// Parquet files can also be used to create a temporary view and then used in SQL statementsparquetFileDF.createOrReplaceTempView("parquetFile")valnamesDF=spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")namesDF.map(attributes=>"Name: "+attributes(0)).show()// +------------+// | value|// +------------+// |Name: Justin|// +------------+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Table partitioning （表分区）是在像 Hive 这样的系统中使用的常见的优化方法. 在 partitioned table （分区表）中, 数据通常存储在不同的目录中, partitioning column values encoded （分区列值编码）在每个 partition directory （分区目录）的路径中. Parquet data source （Parquet 数据源）现在可以自动 discover （发现）和 infer （推断）分区信息. 例如, 我们可以使用以下 directory structure （目录结构）将所有以前使用的 population data （人口数据）存储到 partitioned table （分区表）中, 其中有两个额外的列gender和country作为 partitioning columns （分区列）:
path└── to └── table ├── gender=male │ ├── ... │ │ │ ├── country=US │ │ └── data.parquet │ ├── country=CN │ │ └── data.parquet │ └── ... └── gender=female ├── ... │ ├── country=US │ └── data.parquet ├── country=CN │ └── data.parquet └── ...
通过将path/to/table传递给SparkSession.read.parquet或SparkSession.read.load, Spark SQL 将自动从路径中提取 partitioning information （分区信息）. 现在返回的 DataFrame 的 schema （模式）变成:
root|-- name: string (nullable = true)|-- age: long (nullable = true)|-- gender: string (nullable = true)|-- country: string (nullable = true)
请注意, 会自动 inferred （推断） partitioning columns （分区列）的 data types （数据类型）.目前, 支持 numeric data types （数字数据类型）和 string type （字符串类型）.有些用户可能不想自动推断 partitioning columns （分区列）的数据类型.对于这些用例, automatic type inference （自动类型推断）可以由spark.sql.sources.partitionColumnTypeInference.enabled配置, 默认为true.当禁用 type inference （类型推断）时, string type （字符串类型）将用于 partitioning columns （分区列）.
从 Spark 1.6.0 开始, 默认情况下, partition discovery （分区发现）只能找到给定路径下的 partitions （分区）.对于上述示例, 如果用户将path/to/table/gender=male传递给SparkSession.read.parquet或SparkSession.read.load, 则gender将不被视为 partitioning column （分区列）.如果用户需要指定 partition discovery （分区发现）应该开始的基本路径, 则可以在数据源选项中设置basePath.例如, 当path/to/table/gender=male是数据的路径并且用户将basePath设置为path/to/table/,gender将是一个 partitioning column （分区列）.
像 ProtocolBuffer , Avro 和 Thrift 一样, Parquet 也支持 schema evolution （模式演进）. 用户可以从一个 simple schema （简单的架构）开始, 并根据需要逐渐向 schema 添加更多的 columns （列）. 以这种方式, 用户可能会使用不同但相互兼容的 schemas 的 multiple Parquet files （多个 Parquet 文件）. Parquet data source （Parquet 数据源）现在能够自动检测这种情况并 merge （合并）所有这些文件的 schemas .
由于 schema merging （模式合并）是一个 expensive operation （相对昂贵的操作）, 并且在大多数情况下不是必需的, 所以默认情况下从 1.5.0 开始. 你可以按照如下的方式启用它:
读取 Parquet 文件时, 将 data source option （数据源选项）mergeSchema设置为true（如下面的例子所示）, 或
将 global SQL option （全局 SQL 选项）spark.sql.parquet.mergeSchema设置为true.
// This is used to implicitly convert an RDD to a DataFrame.importspark.implicits._// Create a simple DataFrame, store into a partition directoryvalsquaresDF=spark.sparkContext.makeRDD(1to5).map(i=>(i,i*i)).toDF("value","square")squaresDF.write.parquet("data/test_table/key=1")// Create another DataFrame in a new partition directory,// adding a new column and dropping an existing columnvalcubesDF=spark.sparkContext.makeRDD(6to10).map(i=>(i,i*i*i)).toDF("value","cube")cubesDF.write.parquet("data/test_table/key=2")// Read the partitioned tablevalmergedDF=spark.read.option("mergeSchema","true").parquet("data/test_table")mergedDF.printSchema()// The final schema consists of all 3 columns in the Parquet files together// with the partitioning column appeared in the partition directory paths// root// |-- value: int (nullable = true)// |-- square: int (nullable = true)// |-- cube: int (nullable = true)// |-- key: int (nullable = true)
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Hive metastore Parquet table conversion （Hive metastore Parquet table 转换）
当读取和写入 Hive metastore Parquet 表时, Spark SQL 将尝试使用自己的 Parquet support （Parquet 支持）, 而不是 Hive SerDe 来获得更好的性能. 此 behavior （行为）由spark.sql.hive.convertMetastoreParquet配置控制, 默认情况下 turned on （打开）.
Hive/Parquet Schema Reconciliation
从 table schema processing （表格模式处理）的角度来说, Hive 和 Parquet 之间有两个关键的区别.
Hive 不区分大小写, 而 Parquet 不是
Hive 认为所有 columns （列）都可以为空, 而 Parquet 中的可空性是 significant （重要）的.
由于这个原因, 当将 Hive metastore Parquet 表转换为 Spark SQL Parquet 表时, 我们必须调整 metastore schema 与 Parquet schema. reconciliation 规则是:
在两个 schema 中具有 same name （相同名称）的 Fields （字段）必须具有 same data type （相同的数据类型）, 而不管 nullability （可空性）. reconciled field 应具有 Parquet 的数据类型, 以便 nullability （可空性）得到尊重.
reconciled schema （调和模式）正好包含 Hive metastore schema 中定义的那些字段.
只出现在 Parquet schema 中的任何字段将被 dropped （删除）在 reconciled schema 中.
仅在 Hive metastore schema 中出现的任何字段在 reconciled schema 中作为 nullable field （可空字段）添加.
Spark SQL 缓存 Parquet metadata 以获得更好的性能. 当启用 Hive metastore Parquet table conversion （转换）时, 这些 converted tables （转换表）的 metadata （元数据）也被 cached （缓存）. 如果这些表由 Hive 或其他外部工具更新, 则需要手动刷新以确保 consistent metadata （一致的元数据）.
// spark is an existing SparkSessionspark.catalog.refreshTable("my_table")
可以使用SparkSession上的setConf方法或使用 SQL 运行SET key = value命令来完成 Parquet 的配置.
Property Name （参数名称）Default（默认）Meaning（含义）
spark.sql.parquet.binaryAsStringfalse一些其他 Parquet-producing systems （Parquet 生产系统）, 特别是 Impala, Hive 和旧版本的 Spark SQL , 在 writing out （写出） Parquet schema 时, 不区分 binary data （二进制数据）和 strings （字符串）. 该 flag 告诉 Spark SQL 将 binary data （二进制数据）解释为 string （字符串）以提供与这些系统的兼容性.
spark.sql.parquet.int96AsTimestamptrue一些 Parquet-producing systems , 特别是 Impala 和 Hive , 将 Timestamp 存入INT96 . 该 flag 告诉 Spark SQL 将 INT96 数据解析为 timestamp 以提供与这些系统的兼容性.
spark.sql.parquet.cacheMetadatatrue打开 Parquet schema metadata 的缓存. 可以加快查询静态数据.
spark.sql.parquet.compression.codecsnappy在编写 Parquet 文件时设置 compression codec （压缩编解码器）的使用. 可接受的值包括: uncompressed, snappy, gzip, lzo .
spark.sql.parquet.filterPushdowntrue设置为 true 时启用 Parquet filter push-down optimization .
spark.sql.hive.convertMetastoreParquettrue当设置为 false 时, Spark SQL 将使用 Hive SerDe 作为 parquet tables , 而不是内置的支持.
spark.sql.parquet.mergeSchemafalse当为 true 时, Parquet data source （Parquet 数据源） merges （合并）从所有 data files （数据文件）收集的 schemas , 否则如果没有可用的 summary file , 则从 summary file 或 random data file 中挑选 schema .
spark.sql.optimizer.metadataOnlytrue如果为 true , 则启用使用表的 metadata 的 metadata-only query optimization 来生成 partition columns （分区列）而不是 table scans （表扫描）. 当 scanned （扫描）的所有 columns （列）都是 partition columns （分区列）并且 query （查询）具有满足 distinct semantics （不同语义）的 aggregate operator （聚合运算符）时, 它将适用.
Spark SQL 可以 automatically infer （自动推断）JSON dataset 的 schema, 并将其作为Dataset[Row]加载. 这个 conversion （转换）可以在Dataset[String]上使用SparkSession.read.json()来完成, 或 JSON 文件.
请注意, 以a json file提供的文件不是典型的 JSON 文件. 每行必须包含一个 separate （单独的）, self-contained valid （独立的有效的）JSON 对象. 有关更多信息, 请参阅JSON Lines text format, also called newline-delimited JSON.
对于 regular multi-line JSON file （常规的多行 JSON 文件）, 将multiLine选项设置为true.
// Primitive types (Int, String, etc) and Product types (case classes) encoders are// supported by importing this when creating a Dataset.importspark.implicits._// A JSON dataset is pointed to by path.// The path can be either a single text file or a directory storing text filesvalpath="examples/src/main/resources/people.json"valpeopleDF=spark.read.json(path)// The inferred schema can be visualized using the printSchema() methodpeopleDF.printSchema()// root// |-- age: long (nullable = true)// |-- name: string (nullable = true)// Creates a temporary view using the DataFramepeopleDF.createOrReplaceTempView("people")// SQL statements can be run by using the sql methods provided by sparkvalteenagerNamesDF=spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")teenagerNamesDF.show()// +------+// | name|// +------+// |Justin|// +------+// Alternatively, a DataFrame can be created for a JSON dataset represented by// a Dataset[String] storing one JSON object per stringvalotherPeopleDataset=spark.createDataset("""{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}"""::Nil)valotherPeople=spark.read.json(otherPeopleDataset)otherPeople.show()// +---------------+----+// | address|name|// +---------------+----+// |[Columbus,Ohio]| Yin|// +---------------+----+
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
Spark SQL 还支持读取和写入存储在Apache Hive中的数据。 但是，由于 Hive 具有大量依赖关系，因此这些依赖关系不包含在默认 Spark 分发中。 如果在类路径中找到 Hive 依赖项，Spark 将自动加载它们。 请注意，这些 Hive 依赖关系也必须存在于所有工作节点上，因为它们将需要访问 Hive 序列化和反序列化库 (SerDes)，以访问存储在 Hive 中的数据。
通过将hive-site.xml,core-site.xml（用于安全配置）和hdfs-site.xml（用于 HDFS 配置）文件放在conf/中来完成配置。
当使用 Hive 时，必须用 Hive 支持实例化SparkSession，包括连接到持续的 Hive 转移，支持 Hive serdes 和 Hive 用户定义的功能。 没有现有 Hive 部署的用户仍然可以启用 Hive 支持。 当hive-site.xml未配置时，上下文会自动在当前目录中创建metastore_db，并创建由spark.sql.warehouse.dir配置的目录，该目录默认为Spark应用程序当前目录中的spark-warehouse目录 开始了 请注意，自从2.0.0以来，hive-site.xml中的hive.metastore.warehouse.dir属性已被弃用。 而是使用spark.sql.warehouse.dir来指定仓库中数据库的默认位置。 您可能需要向启动 Spark 应用程序的用户授予写权限。å
importjava.io.Fileimportorg.apache.spark.sql.Rowimportorg.apache.spark.sql.SparkSessioncaseclassRecord(key:Int,value:String)// warehouseLocation points to the default location for managed databases and tablesvalwarehouseLocation=newFile("spark-warehouse").getAbsolutePathvalspark=SparkSession.builder().appName("Spark Hive Example").config("spark.sql.warehouse.dir",warehouseLocation).enableHiveSupport().getOrCreate()importspark.implicits._importspark.sqlsql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive")sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")// Queries are expressed in HiveQLsql("SELECT * FROM src").show()// +---+-------+// |key| value|// +---+-------+// |238|val_238|// | 86| val_86|// |311|val_311|// ...// Aggregation queries are also supported.sql("SELECT COUNT(*) FROM src").show()// +--------+// |count(1)|// +--------+// | 500 |// +--------+// The results of SQL queries are themselves DataFrames and support all normal functions.valsqlDF=sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key")// The items in DataFrames are of type Row, which allows you to access each column by ordinal.valstringsDS=sqlDF.map{caseRow(key:Int,value:String)=>s"Key:$key, Value:$value"}stringsDS.show()// +--------------------+// | value|// +--------------------+// |Key: 0, Value: val_0|// |Key: 0, Value: val_0|// |Key: 0, Value: val_0|// ...// You can also use DataFrames to create temporary views within a SparkSession.valrecordsDF=spark.createDataFrame((1to100).map(i=>Record(i,s"val_$i")))recordsDF.createOrReplaceTempView("records")// Queries can then join DataFrame data with data stored in Hive.sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()// +---+------+---+------+// |key| value|key| value|// +---+------+---+------+// | 2| val_2| 2| val_2|// | 4| val_4| 4| val_4|// | 5| val_5| 5| val_5|// ...
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/hive/SparkHiveExample.scala" in the Spark repo.
创建 Hive 表时，需要定义如何 从/向 文件系统 read/write 数据，即 “输入格式” 和 “输出格式”。 您还需要定义该表如何将数据反序列化为行，或将行序列化为数据，即 “serde”。 以下选项可用于指定存储格式 (“serde”, “input format”, “output format”)，例如，CREATE TABLE src(id int) USING hive OPTIONS(fileFormat 'parquet')。 默认情况下，我们将以纯文本形式读取表格文件。 请注意，Hive 存储处理程序在创建表时不受支持，您可以使用 Hive 端的存储处理程序创建一个表，并使用 Spark SQL 来读取它。
Property NameMeaning
fileFormatfileFormat是一种存储格式规范的包，包括 "serde"，"input format" 和 "output format"。 目前我们支持6个文件格式：'sequencefile'，'rcfile'，'orc'，'parquet'，'textfile'和'avro'。
inputFormat, outputFormat这两个选项将相应的 "InputFormat" 和 "OutputFormat" 类的名称指定为字符串文字，例如: `org.apache.hadoop.hive.ql.io.orc.OrcInputFormat`。 这两个选项必须成对出现，如果您已经指定了 "fileFormat" 选项，则无法指定它们。
serde此选项指定 serde 类的名称。 当指定 `fileFormat` 选项时，如果给定的 `fileFormat` 已经包含 serde 的信息，那么不要指定这个选项。 目前的 "sequencefile", "textfile" 和 "rcfile" 不包含 serde 信息，你可以使用这3个文件格式的这个选项。
fieldDelim, escapeDelim, collectionDelim, mapkeyDelim, lineDelim这些选项只能与 "textfile" 文件格式一起使用。它们定义如何将分隔的文件读入行。
使用OPTIONS定义的所有其他属性将被视为 Hive serde 属性。
Spark SQL 的 Hive 支持的最重要的部分之一是与 Hive metastore 进行交互，这使得 Spark SQL 能够访问 Hive 表的元数据。 从 Spark 1.4.0 开始，使用 Spark SQL 的单一二进制构建可以使用下面所述的配置来查询不同版本的 Hive 转移。 请注意，独立于用于与转移点通信的 Hive 版本，内部 Spark SQL 将针对 Hive 1.2.1 进行编译，并使用这些类进行内部执行（serdes，UDF，UDAF等）。
以下选项可用于配置用于检索元数据的 Hive 版本：
属性名称默认值含义
spark.sql.hive.metastore.version1.2.1Hive metastore 版本。 可用选项为0.12.0至1.2.1。
spark.sql.hive.metastore.jarsbuiltin当启用-Phive时，使用 Hive 1.2.1，它与 Spark 程序集捆绑在一起。选择此选项时，spark.sql.hive.metastore.version 必须为1.2.1或未定义。 行家 使用从Maven存储库下载的指定版本的Hive jar。 通常不建议在生产部署中使用此配置。 ***** 应用于实例化 HiveMetastoreClient 的 jar 的位置。该属性可以是三个选项之一:
builtin当启用-Phive时，使用 Hive 1.2.1，它与 Spark 程序集捆绑在一起。选择此选项时，spark.sql.hive.metastore.version必须为1.2.1或未定义。
maven使用从 Maven 存储库下载的指定版本的 Hive jar。通常不建议在生产部署中使用此配置。
JVM 的标准格式的 classpath。 该类路径必须包含所有 Hive 及其依赖项，包括正确版本的 Hadoop。这些罐只需要存在于 driver 程序中，但如果您正在运行在 yarn 集群模式，那么您必须确保它们与应用程序一起打包。
spark.sql.hive.metastore.sharedPrefixescom.mysql.jdbc,
org.postgresql,
com.microsoft.sqlserver,
oracle.jdbc使用逗号分隔的类前缀列表，应使用在 Spark SQL 和特定版本的 Hive 之间共享的类加载器来加载。 一个共享类的示例就是用来访问 Hive metastore 的 JDBC driver。 其它需要共享的类，是需要与已经共享的类进行交互的。 例如，log4j 使用的自定义 appender。
spark.sql.hive.metastore.barrierPrefixes(empty)一个逗号分隔的类前缀列表，应该明确地为 Spark SQL 正在通信的 Hive 的每个版本重新加载。 例如，在通常将被共享的前缀中声明的 Hive UDF （即： �org.apache.spark.*）。
Spark SQL 还包括可以使用 JDBC 从其他数据库读取数据的数据源。此功能应优于使用JdbcRDD。 这是因为结果作为 DataFrame 返回，并且可以轻松地在 Spark SQL 中处理或与其他数据源连接。 JDBC 数据源也更容易从 Java 或 Python 使用，因为它不需要用户提供 ClassTag。（请注意，这不同于 Spark SQL JDBC 服务器，允许其他应用程序使用 Spark SQL 运行查询）。
要开始使用，您需要在 Spark 类路径中包含特定数据库的 JDBC driver 程序。 例如，要从 Spark Shell 连接到 postgres，您将运行以下命令:
bin/spark-shell --driver-class-path postgresql-9.4.1207.jar --jars postgresql-9.4.1207.jar
可以使用 Data Sources API 将来自远程数据库的表作为 DataFrame 或 Spark SQL 临时视图进行加载。 用户可以在数据源选项中指定 JDBC 连接属性。用户和密码通常作为登录数据源的连接属性提供。 除了连接属性外，Spark 还支持以下不区分大小写的选项:
�属性名称含义
url要连接的JDBC URL。 源特定的连接属性可以在URL中指定。 例如jdbc：jdbc:postgresql://localhost/test?user=fred&password=secret
dbtable应该读取的 JDBC 表。请注意，可以使用在SQL查询的FROM子句中有效的任何内容。 例如，您可以使用括号中的子查询代替完整表。
driver用于连接到此 URL 的 JDBC driver 程序的类名。
partitionColumn, lowerBound, upperBound如果指定了这些选项，则必须指定这些选项。 另外，必须指定numPartitions. 他们描述如何从多个 worker 并行读取数据时将表给分区。partitionColumn必须是有问题的表中的数字列。 请注意，lowerBound和upperBound仅用于决定分区的大小，而不是用于过滤表中的行。 因此，表中的所有行将被分区并返回。此选项仅适用于读操作。
numPartitions在表读写中可以用于并行度的最大分区数。这也确定并发JDBC连接的最大数量。 如果要写入的分区数超过此限制，则在写入之前通过调用coalesce(numPartitions)将其减少到此限制。
fetchsizeJDBC 抓取的大小，用于确定每次数据往返传递的行数。 这有利于提升 JDBC driver 的性能，它们的默认值较小（例如: Oracle 是 10 行）。 该选项仅适用于读取操作。
batchsizeJDBC 批处理的大小，用于确定每次数据往返传递的行数。 这有利于提升 JDBC driver 的性能。 该选项仅适用于写操作。默认值为1000.
isolationLevel事务隔离级别，适用于当前连接。 它可以是NONE,READ_COMMITTED,READ_UNCOMMITTED,REPEATABLE_READ, 或SERIALIZABLE之一，对应于 JDBC 连接对象定义的标准事务隔离级别，默认为READ_UNCOMMITTED。 此选项仅适用于写操作。请参考java.sql.Connection中的文档。
truncate这是一个与 JDBC 相关的选项。 启用SaveMode.Overwrite时，此选项会导致 Spark 截断现有表，而不是删除并重新创建。 这可以更有效，并且防止表元数据（例如，索引）被移除。 但是，在某些情况下，例如当新数据具有不同的模式时，它将无法工作。 它默认为false。 此选项仅适用于写操作。
createTableOptions这是一个与JDBC相关的选项。 如果指定，此选项允许在创建表时设置特定于数据库的表和分区选项（例如：CREATE TABLE t (name string) ENGINE=InnoDB.）。此选项仅适用于写操作。
createTableColumnTypes使用数据库列数据类型而不是默认值，创建表时。 数据类型信息应以与 CREATE TABLE 列语法相同的格式指定（例如："name CHAR(64), comments VARCHAR(1024)"）。 指定的类型应该是有效的 spark sql 数据类型。此选项仅适用于写操作。
// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods// Loading data from a JDBC sourcevaljdbcDF=spark.read.format("jdbc").option("url","jdbc:postgresql:dbserver").option("dbtable","schema.tablename").option("user","username").option("password","password").load()valconnectionProperties=newProperties()connectionProperties.put("user","username")connectionProperties.put("password","password")valjdbcDF2=spark.read.jdbc("jdbc:postgresql:dbserver","schema.tablename",connectionProperties)// Saving data to a JDBC sourcejdbcDF.write.format("jdbc").option("url","jdbc:postgresql:dbserver").option("dbtable","schema.tablename").option("user","username").option("password","password").save()jdbcDF2.write.jdbc("jdbc:postgresql:dbserver","schema.tablename",connectionProperties)// Specifying create table column data types on writejdbcDF.write.option("createTableColumnTypes","name CHAR(64), comments VARCHAR(1024)").jdbc("jdbc:postgresql:dbserver","schema.tablename",connectionProperties)
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" in the Spark repo.
JDBC driver 程序类必须对客户端会话和所有执行程序上的原始类加载器可见。 这是因为 Java 的 DriverManager 类执行安全检查，导致它忽略原始类加载器不可见的所有 driver 程序，当打开连接时。一个方便的方法是修改所有工作节点上的compute_classpath.sh 以包含您的 driver 程序 JAR。
一些数据库，例如 H2，将所有名称转换为大写。 您需要使用大写字母来引用 Spark SQL 中的这些名称。
对于某些工作负载，可以通过缓存内存中的数据或打开一些实验选项来提高性能。
Spark SQL 可以通过调用spark.catalog.cacheTable("tableName")或dataFrame.cache()来使用内存中的列格式来缓存表。 然后，Spark SQL 将只扫描所需的列，并将自动调整压缩以最小化内存使用量和 GC 压力。 您可以调用spark.catalog.uncacheTable("tableName")从内存中删除该表。
内存缓存的配置可以使用SparkSession上的setConf方法或使用 SQL 运行SET key=value命令来完成。
属性名称默认含义
spark.sql.inMemoryColumnarStorage.compressedtrue当设置为 true 时，Spark SQL 将根据数据的统计信息为每个列自动选择一个压缩编解码器。
spark.sql.inMemoryColumnarStorage.batchSize10000控制批量的柱状缓存的大小。更大的批量大小可以提高内存利用率和压缩率，但是在缓存数据时会冒出 OOM 风险。
以下选项也可用于调整查询执行的性能。这些选项可能会在将来的版本中被废弃，因为更多的优化是自动执行的。
属性名称默认值含义
spark.sql.files.maxPartitionBytes134217728 (128 MB)在读取文件时，将单个分区打包的最大字节数。
spark.sql.files.openCostInBytes4194304 (4 MB)按照字节数来衡量的打开文件的估计费用可以在同一时间进行扫描。 将多个文件放入分区时使用。最好过度估计，那么具有小文件的分区将比具有较大文件的分区（首先计划的）更快。
spark.sql.broadcastTimeout300广播连接中的广播等待时间超时（秒）
spark.sql.autoBroadcastJoinThreshold10485760 (10 MB)配置执行连接时将广播给所有工作节点的表的最大大小（以字节为单位）。 通过将此值设置为-1可以禁用广播。 请注意，目前的统计信息仅支持 Hive Metastore 表，其中已运行命令ANALYZE TABLE COMPUTE STATISTICS noscan。
spark.sql.shuffle.partitions200Configures the number of partitions to use when shuffling data for joins or aggregations.
Spark SQL 也可以充当使用其 JDBC/ODBC 或命令行界面的分布式查询引擎。 在这种模式下，最终用户或应用程序可以直接与 Spark SQL 交互运行 SQL 查询，而不需要编写任何代码。
这里实现的 Thrift JDBC/ODBC 服务器对应于 Hive 1.2 中的HiveServer2。 您可以使用 Spark 或 Hive 1.2.1 附带的直线脚本测试 JDBC 服务器。
要启动 JDBC/ODBC 服务器，请在 Spark 目录中运行以下命令:
./sbin/start-thriftserver.sh
此脚本接受所有bin/spark-submit命令行选项，以及--hiveconf选项来指定 Hive 属性。 您可以运行./sbin/start-thriftserver.sh --help查看所有可用选项的完整列表。 默认情况下，服务器监听 localhost:10000. 您可以通过环境变量覆盖此行为，即:
exportHIVE_SERVER2_THRIFT_PORT=exportHIVE_SERVER2_THRIFT_BIND_HOST=./sbin/start-thriftserver.sh\--master \...
or system properties:
./sbin/start-thriftserver.sh\--hiveconf hive.server2.thrift.port=\--hiveconf hive.server2.thrift.bind.host=\--master ...
现在，您可以使用 beeline 来测试 Thrift JDBC/ODBC 服务器:
./bin/beeline
使用 beeline 方式连接到 JDBC/ODBC 服务器:
beeline> !connect jdbc:hive2://localhost:10000
Beeline 将要求您输入用户名和密码。 在非安全模式下，只需输入机器上的用户名和空白密码即可。 对于安全模式，请按照beeline 文档中的说明进行操作。
配置Hive是通过将hive-site.xml,core-site.xml和hdfs-site.xml文件放在conf/中完成的。
您也可以使用 Hive 附带的 beeline 脚本。
Thrift JDBC 服务器还支持通过 HTTP 传输发送 thrift RPC 消息。 使用以下设置启用 HTTP 模式作为系统属性或在conf/中的hive-site.xml文件中启用:
hive.server2.transport.mode - Set this to value: http
hive.server2.thrift.http.port - HTTP port number to listen on; default is 10001
hive.server2.http.endpoint - HTTP endpoint; default is cliservice
要测试，请使用 beeline 以 http 模式连接到 JDBC/ODBC 服务器:
beeline> !connect jdbc:hive2://:/?hive.server2.transport.mode=http;hive.server2.thrift.http.path=
Spark SQL CLI 是在本地模式下运行 Hive 转移服务并执行从命令行输入的查询的方便工具。 请注意，Spark SQL CLI 不能与 Thrift JDBC 服务器通信。
要启动 Spark SQL CLI，请在 Spark 目录中运行以下命令:
./bin/spark-sql
配置 Hive 是通过将hive-site.xml,core-site.xml和hdfs-site.xml文件放在conf/中完成的。 您可以运行./bin/spark-sql --help获取所有可用选项的完整列表。
Spark 2.1.1 介绍了一个新的配置 key:spark.sql.hive.caseSensitiveInferenceMode. 它的默认设置是NEVER_INFER, 其行为与 2.1.0 保持一致. 但是，Spark 2.2.0 将此设置的默认值更改为 “INFER_AND_SAVE”，以恢复与底层文件 schema（模式）具有大小写混合的列名称的 Hive metastore 表的兼容性。使用INFER_AND_SAVE配置的 value, 在第一次访问 Spark 将对其尚未保存推测 schema（模式）的任何 Hive metastore 表执行 schema inference（模式推断）. 请注意，对于具有数千个 partitions（分区）的表，模式推断可能是非常耗时的操作。如果不兼容大小写混合的列名，您可以安全地将spark.sql.hive.caseSensitiveInferenceMode设置为NEVER_INFER，以避免模式推断的初始开销。请注意，使用新的默认INFER_AND_SAVE设置，模式推理的结果被保存为 metastore key 以供将来使用。因此，初始模式推断仅发生在表的第一次访问。
Datasource tables（数据源表）现在存储了 Hive metastore 中的 partition metadata（分区元数据）. 这意味着诸如ALTER TABLE PARTITION ... SET LOCATION这样的 Hive DDLs 现在使用 Datasource API 可用于创建 tables（表）.
遗留的数据源表可以通过MSCK REPAIR TABLE命令迁移到这种格式。建议迁移遗留表利用 Hive DDL 的支持和提供的计划性能。
要确定表是否已迁移，当在表上发出DESCRIBE FORMATTED命令时请查找PartitionProvider: Catalog属性.
Datasource tables（数据源表）的INSERT OVERWRITE TABLE ... PARTITION ...行为的更改。
在以前的 Spark 版本中，INSERT OVERWRITE覆盖了整个 Datasource table，即使给出一个指定的 partition. 现在只有匹配规范的 partition 被覆盖。
请注意，这仍然与 Hive 表的行为不同，Hive 表仅覆盖与新插入数据重叠的分区。
SparkSession现在是 Spark 新的切入点, 它替代了老的SQLContext和HiveContext。注意 : 为了向下兼容，老的 SQLContext 和 HiveContext 仍然保留。可以从SparkSession获取一个新的catalog接口 — 现有的访问数据库和表的 API，如listTables，createExternalTable，dropTempView，cacheTable都被移到该接口。
Dataset API 和 DataFrame API 进行了统一。在 Scala 中，DataFrame变成了Dataset[Row]类型的一个别名，而 Java API 使用者必须将DataFrame替换成Dataset。Dataset 类既提供了强类型转换操作（如map，filter以及groupByKey）也提供了非强类型转换操作（如select和groupBy）。由于编译期的类型安全不是 Python 和 R 语言的一个特性，Dataset 的概念并不适用于这些语言的 API。相反，DataFrame仍然是最基本的编程抽象, 就类似于这些语言中单节点 data frame 的概念。
Dataset 和 DataFrame API 中 unionAll 已经过时并且由union替代。
Dataset 和 DataFrame API 中 explode 已经过时，作为选择，可以结合 select 或 flatMap 使用functions.explode()。
Dataset 和 DataFrame API 中registerTempTable已经过时并且由createOrReplaceTempView替代。
对 Hive tablesCREATE TABLE ... LOCATION行为的更改.
从 Spark 2.0 开始，CREATE TABLE ... LOCATION与CREATE EXTERNAL TABLE ... LOCATION是相同的，以防止意外丢弃用户提供的 locations（位置）中的现有数据。这意味着，在用户指定位置的 Spark SQL 中创建的 Hive 表始终是 Hive 外部表。删除外部表将不会删除数据。 用户不能指定 Hive managed tables（管理表）的位置. 请注意，这与Hive行为不同。
因此，这些表上的 “DROP TABLE” 语句不会删除数据。
从 Spark 1.6 开始，默认情况下服务器在多 session（会话）模式下运行。这意味着每个 JDBC/ODBC 连接拥有一份自己的 SQL 配置和临时函数注册。缓存表仍在并共享。如果您希望以旧的单会话模式运行 Thrift server，请设置选项spark.sql.hive.thriftServer.singleSession为true。您既可以将此选项添加到spark-defaults.conf，或者通过--conf将它传递给start-thriftserver.sh。
./sbin/start-thriftserver.sh\--conf spark.sql.hive.thriftServer.singleSession=true\...
从 1.6.1 开始，在 sparkR 中 withColumn 方法支持添加一个新列或更换 DataFrame 同名的现有列。
从 Spark 1.6 开始，LongType 强制转换为 TimestampType 期望是秒，而不是微秒。这种更改是为了匹配 Hive 1.2 的行为，以便从 numeric（数值）类型进行更一致的类型转换到 TimestampType。更多详情请参阅SPARK-11724。
使用手动管理的内存优化执行，现在是默认启用的，以及代码生成表达式求值。这些功能既可以通过设置spark.sql.tungsten.enabled为false来禁止使用。
Parquet 的模式合并默认情况下不再启用。它可以通过设置spark.sql.parquet.mergeSchema到true以重新启用。
字符串在 Python 列的 columns（列）现在支持使用点（.）来限定列或访问嵌套值。例如df['table.column.nestedField']。但是，这意味着如果你的列名中包含任何圆点，你现在必须避免使用反引号（如table.column.with.dots.nested）。
在内存中的列存储分区修剪默认是开启的。它可以通过设置spark.sql.inMemoryColumnarStorage.partitionPruning为false来禁用。
无限精度的小数列不再支持，而不是 Spark SQL 最大精度为 38 。当从BigDecimal对象推断模式时，现在使用（38，18）。在 DDL 没有指定精度时，则默认保留Decimal(10, 0)。
时间戳现在存储在 1 微秒的精度，而不是 1 纳秒的。
在 sql 语句中，floating point（浮点数）现在解析为 decimal。HiveQL 解析保持不变。
SQL / DataFrame 函数的规范名称现在是小写（例如 sum vs SUM）。
JSON 数据源不会自动加载由其他应用程序（未通过 Spark SQL 插入到数据集的文件）创建的新文件。对于 JSON 持久表（即表的元数据存储在 Hive Metastore），用户可以使用REFRESH TABLESQL 命令或HiveContext的refreshTable方法，把那些新文件列入到表中。对于代表一个 JSON dataset 的 DataFrame，用户需要重新创建 DataFrame，同时 DataFrame 中将包括新的文件。
PySpark 中 DataFrame 的 withColumn 方法支持添加新的列或替换现有的同名列。
DataFrame data reader/writer interface
基于用户反馈，我们创建了一个新的更流畅的 API，用于读取 (SQLContext.read) 中的数据并写入数据 (DataFrame.write), 并且旧的 API 将过时（例如，SQLContext.parquetFile,SQLContext.jsonFile）.
针对SQLContext.read(Scala,Java,Python) 和DataFrame.write(Scala,Java,Python) 的更多细节，请看 API 文档.
DataFrame.groupBy 保留 grouping columns（分组的列）
根据用户的反馈， 我们更改了DataFrame.groupBy().agg()的默认行为以保留DataFrame结果中的 grouping columns（分组列）. 为了在 1.3 中保持该行为，请设置spark.sql.retainGroupColumns为false.
// In 1.3.x, in order for the grouping column "department" to show up,// it must be included explicitly as part of the agg function call.df.groupBy("department").agg($"department",max("age"),sum("expense"))// In 1.4+, grouping column "department" is included automatically.df.groupBy("department").agg(max("age"),sum("expense"))// Revert to 1.3 behavior (not retaining grouping column) by:sqlContext.setConf("spark.sql.retainGroupColumns","false")
之前 1.4 版本中，DataFrame.withColumn() 只支持添加列。该列将始终在 DateFrame 结果中被加入作为新的列，即使现有的列可能存在相同的名称。从 1.4 版本开始，DataFrame.withColumn() 支持添加与所有现有列的名称不同的列或替换现有的同名列。
请注意，这一变化仅适用于 Scala API，并不适用于 PySpark 和 SparkR。
在 Spark 1.3 中，我们从 Spark SQL 中删除了 “Alpha” 的标签，作为一部分已经清理过的可用的 API 。从 Spark 1.3 版本以上，Spark SQL 将提供在 1.X 系列的其他版本的二进制兼容性。这种兼容性保证不包括被明确标记为不稳定的（即 DeveloperApi 类或 Experimental） API。
升级到 Spark SQL 1.3 版本时，用户会发现最大的变化是，SchemaRDD已更名为DataFrame。这主要是因为 DataFrames 不再从 RDD 直接继承，而是由 RDDS 自己来实现这些功能。DataFrames 仍然可以通过调用.rdd方法转换为 RDDS 。
在 Scala 中，有一个从SchemaRDD到DataFrame类型别名，可以为一些情况提供源代码兼容性。它仍然建议用户更新他们的代码以使用DataFrame来代替。Java 和 Python 用户需要更新他们的代码。
此前 Spark 1.3 有单独的Java兼容类（JavaSQLContext和JavaSchemaRDD），借鉴于 Scala API。在 Spark 1.3 中，Java API 和 Scala API 已经统一。两种语言的用户可以使用SQLContext和DataFrame。一般来说论文类尝试使用两种语言的共有类型（如Array替代了一些特定集合）。在某些情况下不通用的类型情况下，（例如，passing in closures 或 Maps）使用函数重载代替。
此外，该 Java 的特定类型的 API 已被删除。Scala 和 Java 的用户可以使用存在于org.apache.spark.sql.types类来描述编程模式。
许多 Spark 1.3 版本以前的代码示例都以import sqlContext._开始，这提供了从 sqlContext 范围的所有功能。在 Spark 1.3 中，我们移除了从RDDs 到DateFrame再到SQLContext内部对象的隐式转换。用户现在应该写成import sqlContext.implicits._.
此外，隐式转换现在只能使用方法toDF来增加由Product（即 case classes or tuples）构成的RDD，而不是自动应用。
当使用 DSL 内部的函数时（现在使用DataFrameAPI 来替换）, 用户习惯导入org.apache.spark.sql.catalyst.dsl. 相反，应该使用公共的 dataframe 函数 API:import org.apache.spark.sql.functions._.
针对 DataType 删除在 org.apache.spark.sql 包中的一些类型别名（仅限于 Scala）
Spark 1.3 移除存在于基本 SQL 包的DataType类型别名。开发人员应改为导入类org.apache.spark.sql.types。
UDF 注册迁移到sqlContext.udf中 (Java & Scala)
用于注册 UDF 的函数，不管是 DataFrame DSL 还是 SQL 中用到的，都被迁移到SQLContext中的 udf 对象中。
sqlContext.udf.register("strLen",(s:String)=>s.length())
Python UDF 注册保持不变。
Python DataTypes 不再是 Singletons（单例的）
在 Python 中使用 DataTypes 时，你需要先构造它们（如：StringType()），而不是引用一个单例对象。
Spark SQL 在设计时就考虑到了和 Hive metastore，SerDes 以及 UDF 之间的兼容性。目前 Hive SerDes 和 UDF 都是基于 Hive 1.2.1 版本，并且Spark SQL 可以连接到不同版本的Hive metastore（从 0.12.0 到 1.2.1，可以参考与不同版本的 Hive Metastore 交互）
Spark SQL Thrift JDBC server 采用了开箱即用的设计以兼容已有的 Hive 安装版本。你不需要修改现有的 Hive Metastore , 或者改变数据的位置和表的分区。
Spark SQL 支持绝大部分的 Hive 功能，如:
Hive query（查询）语句, 包括:
SELECT
GROUP BY
ORDER BY
CLUSTER BY
SORT BY
所有 Hive 操作, 包括:
关系运算符 (=,⇔,==,<>,<,>,>=,<=, 等等)
算术运算符 (+,-,*,/,%, 等等)
逻辑运算符 (AND,&&,OR,||, 等等)
复杂类型的构造
数学函数 (sign,ln,cos, 等等)
String 函数 (instr,length,printf, 等等)
用户定义函数 (UDF)
用户定义聚合函数 (UDAF)
用户定义 serialization formats (SerDes)
窗口函数
Joins
JOIN
{LEFT|RIGHT|FULL} OUTER JOIN
LEFT SEMI JOIN
CROSS JOIN
Unions
Sub-queries（子查询）
SELECT col FROM ( SELECT a + b AS col from t1) t2
Sampling
Explain
Partitioned tables including dynamic partition insertion
View
所有的 Hive DDL 函数, 包括:
CREATE TABLE
CREATE TABLE AS SELECT
ALTER TABLE
大部分的 Hive Data types（数据类型）, 包括:
TINYINT
SMALLINT
INT
BIGINT
BOOLEAN
FLOAT
DOUBLE
STRING
BINARY
TIMESTAMP
DATE
ARRAY<>
MAP<>
STRUCT<>
以下是目前还不支持的 Hive 函数列表。在 Hive 部署中这些功能大部分都用不到。
主要的 Hive 功能
Tables 使用 buckets 的 Tables: bucket 是 Hive table partition 中的 hash partitioning. Spark SQL 还不支持 buckets.
Esoteric Hive 功能
UNION类型
Unique join
Column 统计信息的收集: Spark SQL does not piggyback scans to collect column statistics at the moment and only supports populating the sizeInBytes field of the hive metastore.
Hive Input/Output Formats
File format for CLI: For results showing back to the CLI, Spark SQL only supports TextOutputFormat.
Hadoop archive
Hive 优化
有少数 Hive 优化还没有包含在 Spark 中。其中一些（比如 indexes 索引）由于 Spark SQL 的这种内存计算模型而显得不那么重要。另外一些在 Spark SQL 未来的版本中会持续跟踪。
Block 级别的 bitmap indexes 和虚拟 columns (用于构建 indexes)
自动为 join 和 groupBy 计算 reducer 个数 : 目前在 Spark SQL 中, 你需要使用 “SET spark.sql.shuffle.partitions=[num_tasks];” 来控制 post-shuffle 的并行度.
仅 Meta-data 的 query: 对于只使用 metadata 就能回答的查询，Spark SQL 仍然会启动计算结果的任务.
Skew data flag: Spark SQL 不遵循 Hive 中 skew 数据的标记.
STREAMTABLEhint in join: Spark SQL 不遵循STREAMTABLEhint.
对于查询结果合并多个小文件: 如果输出的结果包括多个小文件, Hive 可以可选的合并小文件到一些大文件中去，以避免溢出 HDFS metadata. Spark SQL 还不支持这样.
Spark SQL 和 DataFrames 支持下面的数据类型:
Numeric types
ByteType: Represents 1-byte signed integer numbers. The range of numbers is from-128to127.
ShortType: Represents 2-byte signed integer numbers. The range of numbers is from-32768to32767.
IntegerType: Represents 4-byte signed integer numbers. The range of numbers is from-2147483648to2147483647.
LongType: Represents 8-byte signed integer numbers. The range of numbers is from-9223372036854775808to9223372036854775807.
FloatType: Represents 4-byte single-precision floating point numbers.
DoubleType: Represents 8-byte double-precision floating point numbers.
DecimalType: Represents arbitrary-precision signed decimal numbers. Backed internally byjava.math.BigDecimal. ABigDecimalconsists of an arbitrary precision integer unscaled value and a 32-bit integer scale.
String type
StringType: Represents character string values.
Binary type
BinaryType: Represents byte sequence values.
Boolean type
BooleanType: Represents boolean values.
Datetime type
TimestampType: Represents values comprising values of fields year, month, day, hour, minute, and second.
DateType: Represents values comprising values of fields year, month, day.
Complex types
ArrayType(elementType, containsNull): Represents values comprising a sequence of elements with the type ofelementType.containsNullis used to indicate if elements in aArrayTypevalue can havenullvalues.
MapType(keyType, valueType, valueContainsNull): Represents values comprising a set of key-value pairs. The data type of keys are described bykeyTypeand the data type of values are described byvalueType. For aMapTypevalue, keys are not allowed to havenullvalues.valueContainsNullis used to indicate if values of aMapTypevalue can havenullvalues.
StructType(fields): Represents values with the structure described by a sequence ofStructFields (fields).
StructField(name, dataType, nullable): Represents a field in aStructType. The name of a field is indicated byname. The data type of a field is indicated bydataType.nullableis used to indicate if values of this fields can havenullvalues.
Spark SQL 的所有数据类型都在包org.apache.spark.sql.types中. 你可以用下示例示例来访问它们.
importorg.apache.spark.sql.types._
Find full example code at "examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" in the Spark repo.
Data type（数据类型）Scala 中的 Value 类型访问或创建数据类型的 API
ByteTypeByteByteType
ShortTypeShortShortType
IntegerTypeIntIntegerType
LongTypeLongLongType
FloatTypeFloatFloatType
DoubleTypeDoubleDoubleType
DecimalTypejava.math.BigDecimalDecimalType
StringTypeStringStringType
BinaryTypeArray[Byte]BinaryType
BooleanTypeBooleanBooleanType
TimestampTypejava.sql.TimestampTimestampType
DateTypejava.sql.DateDateType
ArrayTypescala.collection.SeqArrayType(elementType, [containsNull])
Note（注意）:containsNull的默认值是true.
MapTypescala.collection.MapMapType(keyType,valueType, [valueContainsNull])
Note（注意）:valueContainsNull的默认值是true.
StructTypeorg.apache.spark.sql.RowStructType(fields)
Note（注意）:fields是 StructFields 的 Seq. 所有, 两个 fields 拥有相同的名称是不被允许的.
StructField该 field（字段）数据类型的 Scala 中的 value 类型 (例如, 数据类型为 IntegerType 的 StructField 是 Int)StructField(name,dataType, [nullable])
Note:nullable的默认值是true.
当处理一些不符合标准浮点数语义的float或double类型时，对于 Not-a-Number(NaN) 需要做一些特殊处理. 具体如下:
NaN = NaN 返回 true.
在 aggregations（聚合）操作中，所有的 NaN values 将被分到同一个组中.
在 join key 中 NaN 可以当做一个普通的值.
NaN 值在升序排序中排到最后，比任何其他数值都大.
原文地址: http://spark.apachecn.org/docs/cn/2.2.0/sql-programming-guide.html
网页地址: http://spark.apachecn.org/
github: https://github.com/apachecn/spark-doc-zh（觉得不错麻烦给个 Star，谢谢！~）
