1，JSON数据集

Spark SQL可以自动推断JSON数据集的模式，并将其作为一个Dataset[Row]。这个转换可以SparkSession.read.json()在一个Dataset[String]或者一个JSON文件上完成。

请注意，作为json文件提供的文件不是典型的JSON文件。每行必须包含一个单独的，独立的有效JSON对象。有关更多信息，请参阅 JSON行文本格式，也称为换行符分隔的JSON。

对于常规的多行JSON文件，请将该multiLine选项设置为true。例如下面的例子：

private def runJsonDatasetExample(spark: SparkSession): Unit = {

import spark.implicits.\_

//创建数据集时，通过导入这些  
//元素可以支持原始类型（Int，String等）和Product类型（case类）编码器。import  spark.implicits.\_

// JSON数据集是通过路径指向的。  
// 路径可以是单个文本文件，也可以是存放文本文件的目录

val path = "examples/src/main/resources/people.json"  
val peopleDF = spark.read.json(path)

//推断的模式可以使用printSchema（）方法  
peopleDF.printSchema()  
// root  
//  |-- age: long (nullable = true)  
//  |-- name: string (nullable = true)

//使用DataFrame  
peopleDF.createOrReplaceTempView("people")

// SQL语句可以使用spark  
val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")  
teenagerNamesDF.show()  
// +------+  
// |  name|  
// +------+  
// |Justin|  
// +------+

//或者，也可以为表示的JSON数据集创建一个DataFrame  
//数据集\[String\]每个字符串  
val otherPeopleDataset = spark.createDataset(  
  """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)  
val otherPeople = spark.read.json(otherPeopleDataset)  
otherPeople.show()  
// +---------------+----+  
// |        address|name|  
// +---------------+----+  
// |\[Columbus,Ohio\]| Yin|  
// +---------------+----+  

}

2，Hive 表

1） Spark SQL也支持读写存储在Apache Hive中的数据。但是，由于Hive具有大量依赖项，因此这些依赖项不包含在默认的Spark分发中。如果可以在类路径上找到Hive依赖关系，则Spark将自动加载它们。请注意，这些Hive依赖项也必须存在于所有工作节点上，因为它们需要访问Hive序列化和反序列化库（SerDes）才能访问存储在Hive中的数据。

       配置Hive是通过放置你的hive-site.xml，core-site.xml（用于安全配置）和hdfs-site.xml（用于HDFS配置）文件来完成的conf/。

       使用Hive时，必须SparkSession使用Hive支持进行实例化，包括连接到持续的Hive Metastore，支持Hive serdes和Hive用户定义的函数。没有现有Hive部署的用户仍然可以启用Hive支持。当未配置时hive-site.xml，上下文metastore_db在当前目录中自动创建，并创建一个目录spark.sql.warehouse.dir，该目录默认为 spark-warehouseSpark应用程序启动的当前目录中的目录。请注意，自Spark 2.0.0以来，该hive.metastore.warehouse.dir属性hive-site.xml已被弃用。而是使用spark.sql.warehouse.dir指定仓库中数据库的默认位置。您可能需要向启动Spark应用程序的用户授予写权限。

import java.io.File

import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession

object SparkHiveExample {

case class Record(key: Int, value: String)

def main(args: Array[String]) {
// warehouseLocation指向托管数据库的默认位置，表
val warehouseLocation = new File("spark-warehouse").getAbsolutePath

val spark = SparkSession  
  .builder()  
  .appName("Spark Hive Example")  
  .config("spark.sql.warehouse.dir", warehouseLocation)  
  .enableHiveSupport()  
  .getOrCreate()

import spark.implicits.\_  
import spark.sql

sql("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")

//查询以HiveQL  
sql("SELECT \* FROM src").show()  
// +---+-------+  
// |key|  value|  
// +---+-------+  
// |238|val\_238|  
// | 86| val\_86|  
// |311|val\_311|  
// ...

//聚合查询也被支持。  
sql("SELECT COUNT(\*) FROM src").show()  
// +--------+  
// |count(1)|  
// +--------+  
// |    500 |  
// +--------+

// SQL查询的结果本身就是DataFrame，并且支持所有正常的函数。  
val sqlDF = sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key")

// DataFrames中的项目类型为Row，它允许您通过序号访问每个列。  
val stringsDS = sqlDF.map {  
  case Row(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|  
// ...

//您也可以使用DataFrame在SparkSession中创建临时视图。  
val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i, s"val\_$i")))  
recordsDF.createOrReplaceTempView("records")

//查询然后可以将DataFrame数据与存储在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|  
// ...  
spark.stop()  

}
}

2） 指定Hive表格的存储格式

      当你创建一个Hive表时，你需要定义这个表应该如何从/向文件系统读/写数据，即“输入格式”和“输出格式”。您还需要定义这个表应该如何将数据反序列化为行，或者将行序列化为数据，即“serde”。以下选项可用于指定存储格式（“serde”，“输入格式”，“输出格式”），例如CREATE TABLE src(id int) USING hive OPTIONS(fileFormat 'parquet')。默认情况下，我们将以纯文本形式读取表格文件。请注意，Hive存储处理程序在创建表时不受支持，您可以使用Hive端的存储处理程序创建一个表，然后使用Spark
SQL来读取它。

属性名称

含义

fileFormat

fileFormat是一种存储格式规范包，包括“serde”，“输入格式”和“输出格式”。目前我们支持6个fileFormats：'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个fileFormats。

fieldDelim, escapeDelim, collectionDelim, mapkeyDelim, lineDelim

这些选项只能与“textfile”fileFormat一起使用。他们定义了如何将分隔文件读入行。

定义的所有其他属性OPTIONS将被视为Hive serde属性。

3） 与不同版本的Hive Metastore交互

Spark SQL的Hive支持最重要的部分之一是与Hive Metastore进行交互，这使Spark SQL可以访问Hive表的元数据。从Spark 1.4.0开始，使用下面描述的配置，可以使用Spark SQL的单个二进制版本查询不同版本的Hive metastore。请注意，独立于用于与Metastore对话的Hive版本，内部Spark SQL将针对Hive 1.2.1进行编译，并将这些类用于内部执行（serdes，UDF，UDAF等）。

以下选项可用于配置用于检索元数据的Hive版本：

属性名称

默认

含义

spark.sql.hive.metastore.version

      1.2.1

hive  Metastore版本。可用的选项是0.12.0通过1.2.1。

spark.sql.hive.metastore.jars

       builtin

应该用来实例化HiveMetastoreClient的罐子的位置。该属性可以是以下三个选项之一：

1，builtin

-Phive启用 时，使用与Spark程序集捆绑在一起的Hive 1.2.1 。选择此选项时，spark.sql.hive.metastore.version

必须是1.2.1或者没有定义。

2，maven

使用从Maven存储库下载的指定版本的Hive jar。通常不建议将此配置用于生产部署。

3，JVM标准格式的类路径。这个类路径必须包含所有Hive及其依赖项，包括正确版本的Hadoop。

这些瓶子只需要在驱动程序中出现，但是如果您正在以纱线群集模式运行，则必须确保它们与您的应用程序一起打包。

spark.sql.hive.metastore.sharedPrefixes

com.mysql.jdbc,

org.postgresql,

com.microsoft.sqlserver,

oracle.jdbc

应该使用在Spark SQL和特定版本的Hive之间共享的类加载程序加载的类前缀的逗号分隔列表。

应该共享的类的示例是需要与Metastore对话的JDBC驱动程序。其他需要共享的类是那些与已

经共享的类进行交互的类。例如，由log4j使用的自定义appender。

spark.sql.hive.metastore.barrierPrefixes

(empty)

一个以逗号分隔的类前缀列表，应该针对Spark SQL正在与之进行通信的每个Hive版本显式

重新加载。例如，Hive UDF声明在通常会被共享（即org.apache.spark.*）的前缀中。

2，JDBC到其他数据库

1） Spark SQL还包含一个可以使用JDBC从其他数据库读取数据的数据源。这个功能应该比使用JdbcRDD更受欢迎。这是因为结果作为DataFrame返回，并且可以轻松地在Spark SQL中处理或者与其他数据源结合使用。JDBC数据源也更容易从Java或Python使用，因为它不需要用户提供ClassTag。（请注意，这与Spark SQL JDBC服务器不同，后者允许其他应用程序使用Spark SQL运行查询）。

       要开始，您将需要在Spark类路径中为您的特定数据库包含JDBC驱动程序。例如，要从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连接属性。 user和password通常用于登录到数据源提供为连接属性。除了连接属性之外，Spark还支持以下不区分大小写的选项：

属性名称

含义

url

要连接到的JDBC URL。源特定的连接属性可以在URL中指定。例如，jdbc:postgresql://localhost/test?user=fred&password=secret

dbtable

应该读取的JDBC表。请注意，FROM可以使用在SQL查询的子句中有效的任何内容。例如，而不是一个完整的表，你也可以在括号中使用子查询。

driver

用于连接到此URL的JDBC驱动程序的类名。

partitionColumn, lowerBound, upperBound

如果指定了这些选项，则必须指定这些选项。另外， numPartitions必须指定。他们描述了如何从多个工作人员平行读取时对表格进行分区。 partitionColumn必须是相关表格中的数字列。请注意，lowerBound和upperBound只是用来决定分区步幅，而不是在表中过滤行。所以表中的所有行都将被分区并返回。这个选项只适用于阅读。

numPartitions

表格读取和写入中可用于并行的分区的最大数目。这也决定了并发JDBC连接的最大数量。如果要写入的分区数量超过此限制，则coalesce(numPartitions)在写入之前调用它，将其减少到此限制。

fetchsize

JDBC提取大小，它决定每次往返取多少行。这可以帮助默认为低读取大小的JDBC驱动程序（例如，具有10行的Oracle）执行性能。这个选项只适用于阅读。

batchsize

JDBC批量大小，用于确定每次往返要插入多少行。这可以帮助JDBC驱动程序的性能。这个选项只适用于写作。它默认为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数据类型，该选项仅适用于写入。

例子如下：

private def runJdbcDatasetExample(spark: SparkSession): Unit = {
//注意：可以通过load / save或jdbc方法来实现JDBC加载和保存
//从JDBC源加载数据
val jdbcDF = spark.read
.format("jdbc")
.option("url", "jdbc:postgresql:dbserver")
.option("dbtable", "schema.tablename")
.option("user", "username")
.option("password", "password")
.load()

val connectionProperties = new Properties()  
connectionProperties.put("user", "username")  
connectionProperties.put("password", "password")  
val jdbcDF2 = spark.read  
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)  
//将数据保存到JDBC源  
connectionProperties.put("customSchema", "id DECIMAL(38, 0), name STRING")  
val jdbcDF3 = spark.read  
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

//在写入  
jdbcDF.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)

jdbcDF.write  
  .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")  
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

}

2） 可能遇到的异常和故障排除

JDBC驱动程序类必须对客户端会话和所有执行者上的原始类加载器可见。这是因为Java的DriverManager类执行了一个安全检查，导致它忽略了当打开一个连接时，原始类加载器不可见的所有驱动程序。一个方便的方法是修改所有工作节点上的compute_classpath.sh以包含驱动程序JAR。

       某些数据库（如H2）将所有名称转换为大写。您需要使用大写字母来引用Spark SQL中的这些名称。