• 概述
• 数据源的加载&保存
  • 数据源加载
  • 数据源保存
    • 保存模式
    • 存储源
• 内置数据源
  • parquet
  • Orc
  • Json
  • Hive
  • JDBC

概述

Spark-SQL是通过DataFrame接口对各种数据源进行操作

• 各种数据源的加载
• 数据的转换
• 数据源的保存
• 注册临时视图,来允许SQL的形式直接对临时视图进行操作等

数据源的加载&保存

数据源加载

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

//默认方式为parquet  
spark.read.load("D:\\data\\users.parquet").show()
//加载非parquet的数据源,需要设置format
//对内置数据源,可以使用短名称.对外部数据源,必须使用全名称(包名+类名)
//内置数据源有 parquet,txt,json,orc,csv,libsvm
spark.read.format("csv").load("D:\\data\\people.csv").show()

//以SQL形式读取文件
spark.sql("select * from json.`D:\\data\\sku.json`").show()

数据源保存

保存模式

注意:
数据保存并非是一个原子性操作.这一点务必注意
比如Append,可能输出过程中断而导致部分数据进入.或者Overwrite,本质是一个删除后新建的操作,这里是有删除后新建失败,导致有新数据无法写入老数据丢失的可能性的

存储源

Spark 支持两种类型的存储源 : 基于文件的存储源(file-based) 和 基于表的存储源(table-based)

**基于文件的存储源(file-based) **

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

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

基于Hive表的存储源(hive-table-based)
将数据保存到Hive的表中

• 元数据
  数据保存到Hive的表中,将同时自动保存schema到Hive的MetaStore里
• 内/外表(数据保存时,根据是否指定path来区分内外表)
  当手动设置为其它路径时成为外部表(表删除只删除元数据不删除表数据)
  不设置path则成为内部表(表删除将同时删除元数据和表数据)
  注意:从Spark2.0之后, CREATE TABLE 但指定path依然会成为一个外部表.即唯一标准是是否指定path
• 表分区
  如果表被设置为外部表,默认不会收集分区信息,需要手动同步(msck / add partition)

内置数据源

parquet

概述
Parquet是一种列式存储格式,无法直接阅读.但有非常好的压缩消耗和压缩比.
Parquet是Spark默认和推荐使用的数据格式.在很多方面,Spark都对Parquet有最大支持
在Spark中,出于兼容性的考虑,所有的Columns都将自动转换为可空类型

读写demo

spark.read.format("csv").load("InputPath").write.format("parquet").save("OutputPath") 

分区发现(Partition Discovery)
在Hive之类中,分区是以目录的形式存在.分区键本身是目录的一部分.但分区信息是不会自动的被Hive发现
而在Parquet中,Spark可以自动的发现分区和推断数据类型.推断的依据就是目录
比如:
一个存放订单信息的目录,以类似这样的格式 order/time=xxxx-MM-dd}/area=xx/xxxx.parquet
传入路径(/order),读取将自动增加两列time和area作为分区列,并自动在实际使用中应用分区信息

注意:

• 对Parquet的分区发现字段现仅支持数字型和字符串型
• spark.sql.sources.partitionColumnTypeInference.enabled(默认为true)设置
  可以关闭分区发现的数据类型推断

元数据合并(schema merging)
实际工作中,parquet数据文件很可能是由一个简单的逐渐变得复杂(随着业务的深入不断增加字段)
但因为parquet是自存储的,所以需要有一种途径来保证对历史数据的兼容. 这就是模式演进(schema evolution),而实现的手段就是元数据合并

注意:
模式合并,是一个昂贵的操作,在大多数的情况,都不会也不应该使用它,所以模式是默认关闭的

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

• 临时
  在Option中设置mergeSchema=true
  spark.read.option(“mergeSchema”, “true”)
• 永久
  配置文件配置 spark.sql.parquet.mergeSchema 设置为 true

parquet 与 Hive MetaStore
在Spark读取和写入Hive中的Parquet存储格式表时,将使用Spark自己的Parquet Support 而不是使用Hive Serde (spark.sql.hive.convertMetastoreParquet 默认开启)

Spark 与 Hive-MetaStore 在处理Parquet-Schema是有一些区别

• Hive-MetaStore不区分大小写,而Spark-Parquet-Schema 区分大小写
• Hive-MetaStore认为列都是可空的,而Spark认为列是否可空必须显式指定

所以在Parquet上,Spark会做出一些自协调如下

• 两个模式中相同名称的字段必须具有相同的数据类型,但不关心是否为空(都为可空)
• 如果列只出现在Spark-Parquet-Schema上但没有出现在Hive-MetaStore上,该列会被删除
• 如果列只出现在Hive-MetaStore而没有出现在Spark-Parquet-Schema上,该列将以可空形式追加
  一个可以参考的最佳实践是:
  对所有列名不采用驼峰命名法,全部使用纯小写下划线分割的下划线命名法

parquet元数据刷新
Spark会缓存Parquet-Schema以提高性能,但如果开启了spark.sql.hive.convertMetastoreParquet,也就是使用Spark自身的Parquet支持,这时来自Hive元数据的信息依然会被缓存,如果此时Hive表在Spark之外被改变,就必须刷新一下Parquet元数据以使用到新的元数据

spark.catalog.refreshTable("my_table")

parquet可用配置

Orc

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

Json

Spark-SQL可以自动推断Json的schema.(Json的schema依然是自存储的)
Sparl-SQL加载将以DataFrame[Row]的形式,并且比较轻松的转换为一个DatasetT

//读取Json数据源,如果某些行缺age属性的,补0
spark.read.json("D:\\data\\people.json").na.fill(0,Seq("age"))

Hive

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

• Hive的库必须在classpath中被找到.因为Hive的库有大量的依赖,而这些依赖不一定是被Spark完全打包的,所以在运行时,必须要能找到这些库包
  注意:这种依赖与找到是针对每个executor而言的,因为executor才是真正的执行者
• 需要在Spark.Conf中拷入 hive-site.xml, core-site.xml（用于安全配置）和 hdfs-site.xml （用于 HDFS 配置）文件
• 需要在SparkSession中启用对Hive的支持:spark.enableHiveSupport()

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

demo

val spark = SparkSession
  .builder()
  .appName("Spark Hive Example")
  .enableHiveSupport()
  .getOrCreate()
  
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")

// Create a Hive managed Parquet table, with HQL syntax instead of the Spark SQL native syntax
// `USING hive`
sql("CREATE TABLE hive_records(key int, value string) STORED AS PARQUET")
// Save DataFrame to the Hive managed table
val df = spark.table("src")
df.write.mode(SaveMode.Overwrite).saveAsTable("hive_records")
// After insertion, the Hive managed table has data now
sql("SELECT * FROM hive_records").show()
// +---+-------+
// |key|  value|
// +---+-------+
// |238|val_238|
// | 86| val_86|
// |311|val_311|
// ...

// Prepare a Parquet data directory
val dataDir = "/tmp/parquet_data"
spark.range(10).write.parquet(dataDir)
// Create a Hive external Parquet table
sql(s"CREATE EXTERNAL TABLE hive_ints(key int) STORED AS PARQUET LOCATION '$dataDir'")
// The Hive external table should already have data
sql("SELECT * FROM hive_ints").show()
// +---+
// |key|
// +---+
// |  0|
// |  1|
// |  2|
// ...

// Turn on flag for Hive Dynamic Partitioning
spark.sqlContext.setConf("hive.exec.dynamic.partition", "true")
spark.sqlContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
// Create a Hive partitioned table using DataFrame API
df.write.partitionBy("key").format("hive").saveAsTable("hive_part_tbl")
// Partitioned column `key` will be moved to the end of the schema.
sql("SELECT * FROM hive_part_tbl").show()
// +-------+---+
// |  value|key|
// +-------+---+
// |val_238|238|
// | val_86| 86|
// |val_311|311|
// ...

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

对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

详情请见官网描述

JDBC

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

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 包含进驱动包

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

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);