7. 数据源

　　Spark-SQL 支持通过Dataframe接口对各种数据源进行操作

　　　　各种数据源的加载&保存

　　　　数据转换(relational transformations)

　　　　注册临时视图(temporary view),来允许SQL的形式直接对临时视图进行操作

　　7.1  数据源加载

　　　　Spark-SQL的默认数据源为parquet(spark.sql.sources.default设置),一些数据源加载的例子如下:　　　　

 /**
        * 加载parquet数据源
        */
        spark.read.load("D:\\data\\users.parquet").show()
        /**
        * 加载非parquet数据源,需要手动设置选项
        *   Spark-SQLd对内置数据源,可以使用短名称.对自定义的数据源需要使用全名称(比如:org.apache.spark.sql.parquet)
        *   Spark-SQL内置数据源:json, parquet, jdbc, orc, libsvm, csv, text
        */
        spark.read.format("csv").load("D:\\data\\people.csv").show()
        /**
        * 以SQL的形式读取文件
        */
        spark.sql("select * from json.`D:\\data\\sku.json`").show()

　　7.2 数据源保存

　　　　7.2.1 保存模式

　　　　　　数据源的保存提供了以下保存的模式

　　　　　　数据源的保存是 非原子性 的,这一点务必要注意.比如以Overwrite形式写入时,数据将先删除后写入(覆盖),这里有数据丢失的可能性的

　　　    7.2.2 存储源

　　　　　　Spark-SQL 可以将数据保存在两种源上: 文件(file-based) 和 持久化表(persistent tables)　

　　　　　　i).文件(file-based)

　　　　　　　　数据保存的格式可以变化,比如将一个csv保存为一个txt的例子

　　　　　　　　spark.read.format(“csv”).load(“InputPath”).write.format(“text”).save(“OutputPath”) 

　　　　　  ii).持久化表(persistent tables)

　　　　　　　　将数据保存到Hive的表中.

　　　　　　　　元数据

　　　　　　　　　　数据保存到Hive的表中,将同时自动保存schema到Hive的MetaStore里

　　　　　　　　内/外表

　　　　　　　　　　数据保存时,根据是否指定path来区分内外表.

　　　　　　　　　　　　当手动设置为其它路径时成为外部表(表删除只删除元数据不删除表数据)

　　　　　　　　　　　　不设置path则成为内部表(表删除将同时删除元数据和表数据)

　　　　　　　　　　　　注意: 从Spark2.0之后, CREATE TABLE 但没有指定path依然会成为一个外部表.即唯一标准是是否指定path

　　　　　　　　表分区

　　　　　　　　　　如果表被设置为外部表,默认不会收集分区信息,需要手动同步(msck / add partition)

　　7.3.内置数据源

　　　　7.3.1 Parquet

　　　　　　  Parquet是一种列式存储格式,无法直接阅读.但有非常好的压缩消耗和压缩比.

　　　　　　  Parquet是Spark默认和推荐使用的数据格式.在很多方面,Spark都对Parquet有最大支持

　　　　　　  在Spark中,出于兼容性的考虑,所有的Columns都将自动转换为可空类型

　　　　　　  7.3.1.1 一个Parquet读写的例子

　　　　　　　　spark.read.format(“csv”).load(“InputPath”).write.format(“text”).save(“OutputPath”) 

　　　　　      7.3.1.2 Parquet的分区发现

　　　　　　　　在Hive之类中,分区是以目录的形式存在.分区键本身是目录的一部分.但此时,分区信息是不会自动的被Hive发现.

　　　　　　　　而在Parquet中,Spark可以自动的发现分区和推断数据类型.推断的依据就是目录

　　　　　　　　比如:一个存放所有订单信息的目录,以类似这样的格式 order/time=xxxx-MM-dd}/area=xx/xxxx.parquet

　　　　　　　　　　 将路径(/order)传入,读取将自动增加两列time和area作为分区列,并自动在实际使用中应用分区信息

　　　　　　　　注意:

　　　　　　　　　　对Parquet的分区发现字段现仅支持数字型和字符串型.

　　　　　　　　spark.sql.sources.partitionColumnTypeInference.enabled(默认为true)设置,可以关闭分区发现的数据类型推断

　　　　　　  7.3.1.3 Parquet的元数据合并(schema merging)

　　　　　　　　 实际工作中,Parquet数据文件很可能是由一个开始比较简单的schema逐渐变得复杂(一开始的业务比较简单,但随着业务的推进会加入越来越多的字段)

　　　　　　　　 因为Parquet的schema是自存储的,所以需要一种途径来保持对历史数据schema的兼容性,这就是模式演进(schema evolution),而实现的手段就是模式合并

　　　　　　　　 模式合并是一个昂贵操作,而在大多数情况下都是不会使用到它,所以模式合并默认是关闭的

　　　　　　　　 有以下两种途径可以打开模式合并

　　　　　　　　　　i).临时=>在Option中设置mergeSchema=true

　　　　　　　　　　　　spark.read.option(“mergeSchema”, “true”)

　　　　　　　　　   ii).永久=> 配置文件配置 spark.sql.parquet.mergeSchema 设置为 true

　　　　　　7.3.1.4 Parquet 与Hive MetaStore

　　　　　　　　在Spark读取和写入Hive的Parquet表时,Spark将使用自己的Parquet Support,而不是Hive SerDe ( spark.sql.hive.convertMetastoreParquet 默认开启)

　　　　　　　　 Spark-Parquet-Schema 与 Hive-MetaStore 是有一些区别的,这些区别体现在:

　　　　　　　　　　Hive-MetaStore不区分大小写,而Spark-Parquet-Schema是区分大小写的.

　　　　　　　　　　Hive-MetaStore认为所有列都是可空的,而Spark-Parquet-Schema认为列是否可空必须指定

　　　　　　　　所以在Spark-Parquet-Schema 与 Hive-MetaStore 会进行一些自协调(reconciled schema),规则如下:

　　　　　　　　　　只出现在Spark-Parquet-Schema中的任何字段会被自动删除

　　　　　　　　　　只出现在Hive-MetaStore中的任何字段以可空字段(nullable field)形式增加

　　　　　　7.3.1.5 Parquet的元数据刷新(Metadata Refreshing)

　　　　　　　　 Parquet的元数据将会被自动缓存以获得更好的性能.

　　　　　　　　 但需要注意的是:Hive metastore Parquet table conversion,这些Hive表的MetaData也会被缓存.

　　　　　　　　 如果这些表在Hive或其它工具层面被改变,Spark无法感知这些改变,必须手工刷新.刷新方式如下:

　　　　　　　　　　spark.catalog.refreshTable("my_table")

　　　　　　7.3.1.6 Parquet的常用配置

　　　　7.3.1 ORC

　　　　　　ORC格式是Spark-2.3之后才支持的数据源格式.它的可配置信息如下:

　　　　7.3.2 JSON

　　　　　　Spark-SQL可以自动推断Json的schema.(Json的schema依然是自存储的)

　　　　　　Spark-SQL加载将以DataFrame[Row]的形式,并且可以非常简单的转为一个Dataset[T](.as[T])

　　　　　　一个简单的例子如下:

　　　　　　　　//读取Json数据源,如果某些行缺age属性的,补0

　　　　　　　　spark.read.json(“D:\\data\\people.json”).na.fill(0,Seq(“age”))

　　　  7.3.3 Hive

　　　　　　7.3.3.1 Hive支持

　　　　　　　　Spark-SQL还支持读取和写入存储在Hive中的数据.Spark-SQL使用Hive,但需要做一些补充配置

　　　　　　　　　　i).Hive的库必须在classpath中被找到.因为Hive的库有大量的依赖,而这些依赖不一定是被Spark完全打包的,所以在运行时,必须要能找到这些库包

　　　　　　　　　　　注意:这种依赖与找到是针对每个executor而言的,因为executor才是真正的执行者.

　　　　　　　  　　 ii).需要在Spark.Conf中拷入 hive-site.xml, core-site.xml（用于安全配置）和 hdfs-site.xml （用于 HDFS 配置）文件

　　　　　　　　　  iii).需要在SparkSession中启用对Hive的支持:spark.enableHiveSupport()

　　　　　　       在实际运行中,不一定需要部署一个完整的Hive系统,比如当hive-site.xml不存在时,会在当前目录创建一个metastore_db用以存储元数据(不推荐如此)

　　　　　　7.3.3.2 Hive的存储格式

　　　　　　　　创建一个Hive表,需要定义读取/写入文件系统的格式,以及数据的行断定义,列断定义等.Spark-Sql读取Hive表时,默认以文本格式读取.

　　　　　　　　一些Hive存储格式相关格式如下:

　　　　　　7.3.3.3 对Hive不支持的地方

　　　　　　　　Spark-SQL,相当于内置了一个Hive引擎,它具有Hive绝大部分的功能,但确实不是完全支持Hive.Spark-SQL不支持的如下:

　　　　　　　　　　Tables with buckets

　　　　　　　　　　UNION  和 Unique join

　　　　　　　　　　Hive的列静态数据统计收集功能

　　　　　　　　　　File format for CLI  Spark-SQL只支持 TextOutputFormat

　　　　　　　　　　对一些UDF函数,Spark-SQL结果也与原生Hive不同

　　　　　　　　　　　　SQRT(n) If n < 0, Hive returns null, Spark SQL returns NaN

　　　　　　　　　　　　ACOS(n) If n < -1 or n > 1, Hive returns null, Spark SQL returns NaN

　　　　　　　　　　　　ASIN(n) If n < -1 or n > 1, Hive returns null, Spark SQL returns NaN

　　　　　　　　　　详情请见官网描述

　　　　7.3.3 JDBC

　　　　　　7.3.3.1 简述

　　　　　　　　Spark-SQL 同样支持以JDBC的形式从其它的关系型数据库读取数据.(此方式优于JdbcRDD,因为DF更容易与处理与其它数据源的交互,比如直接映射为临时视图等等)

　　　　　　　　JDBC-驱动

　　　　　　　　　　JDBC依赖目标数据库驱动包.需要注意的是,这个驱动必须同时对driver端和executor端所有执行程序的类默认加载器可见(DriverManager会忽略不可见的驱动导致执行找不到驱动)

　　　　　　　　　　例如，要从 Spark Shell 连接到 postgres 必须指定驱动包位置

　　　　　　　　　　　　bin/spark-shell --driver-class-path postgresql-9.4.1207.jar --jars postgresql-9.4.1207.jar

　　　　　　　　　　一个比较方便的办法是修改所有工作节点 compute_classpath.sh 包含进驱动包

　　　　　　7.3.3.2 JDBC常用配置

　　　　　　　　可以在数据源选项中指定 JDBC 连接属性。用户 和 密码通常作为登录数据源的连接属性提供。 除了连接属性外，Spark 还支持以下不区分大小写的选项

　　　　　　7.3.3.2 JDBC-Demo

　　　　　　　　/**
    　　　　　　　* 连接RDBMS(SQLServer2008)Demo
    　　　　　　　*   注:必须导入SQLServer驱动包(sqljdbc42.jar),(代码中不能体现,但实际已在项目中添加过了)
    　　　　　　　*/
  　　　　　　　　val df = spark.read.format("jdbc")
                　　　　.option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver")
                　　　　.option("url", "jdbc:sqlserver://127.0.0.1\\Sql2008:1433;DatabaseName=AdapterService")
                　　　　.option("user", "sa")
                　　　　.option("password", "12abAB")
                　　　　.option("dbtable", "dbo.tm_flow")
                　　　　.load()

  　　　　　　　　df.show(10);

　　　　　　7.3.3.3 JDBC-性能调优

　　　　　　　　内存缓存

　　　　　　　　　　Spark-SQL 支持对JDBC结果缓存到内存以提升查询效率

　　　　　　　　　　spark.catalog.cacheTable(“tableName”) 或 dataFrame.cache() 将数据结果以列式存储的形式存放至内存 

　　　　　　　　　　spark.catalog.uncacheTable(“tableName”) 删除缓存

　　　　　　　　广播

　　　　　　　　　　broadcast(spark.table("xxx"))