[Spark]-结构化数据查询之数据源篇

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 保存模式

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

Scala/JavaAny Language描述
SaveMode.ErrorIfExists(默认)error/errorifexists如果保存的数据已存在,将抛出错误
SaveMode.Appendappend如果保存的数据已存在,数据将以追加的形式写入
SaveMode.Overwriteoverwrite如果保存的数据已存在,数据将以覆盖的形式写入
SaveMode.Ignoreignore如果保存的数据已存在,数据将不会写入(也不会抛出错误,即原数据不变,新数据忽略)

      数据源的保存是 非原子性 的,这一点务必要注意.比如以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的常用配置

          

参数名默认值描述
spark.sql.parquet.binaryAsStringfalse

一些其它的基于Parquet系统(比如Hive,Impala,旧版Spark-SQL),在写出Parquet-schema时不区分binary data (二进制数据)和 strings (字符串)

启用这个属性,将保持将二进制数据解释为字符串的兼容性

spark.sql.parquet.int96AsTimestamptrue一些其它的基于Parquet系统(比如Hive,Impala),会将Timestamp写为int96.该属性告知Spark-SQL将int96解析为Timestamp的兼容性
spark.sql.parquet.cacheMetadatatrue指示是否打开parquet的元数据缓存(这可以加快查询静态数据的速度)
spark.sql.parquet.compression.codecsnappy指示写出parquet文件的压缩类型(默认snappy).可选项为:uncompressed, snappy, gzip, lzo
spark.sql.parquet.filterPushdowntrue设置为 true 时启用 (过滤谓词下推,尝试调整算子执行顺序,将写在后面的过滤谓词提前处理以减少查询数据量,提高性能)
spark.sql.hive.convertMetastoreParquettrue面对Hive时,是否启用使用Spark-SQL内置的parquet,设为否则放弃使用内置parquet转而使用Hive serDe
spark.sql.parquet.mergeSchemafalse是否全局启用parquet的元数据合并,设为否则从summary file或random file中随机挑选
spark.sql.optimizer.metadataOnlytrue当设为true时,将使用metadata信息来构建分区列而不是走表扫描.(只是在查询的所有列都是分区列时才有意义,并且此时依然有各种聚合能力)

    7.3.1 ORC

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

参数名默认值描述
spark.sql.orc.implhiveORC的实现名称.可选值hive(使用hive-1.2.1的ORC库),native(或者hive.native,将使用Apache ORC 1.4.1)
spark.sql.orc.enableVectorizedReadertrue是否在本机使用向量化ORC解码.如果为false,会在本机构造一个向量化ORC阅读器,对于上面设置为hive,这个属性将被忽略

    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.xmlcore-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存储格式相关格式如下:

属性名描述
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” 文件格式一起使用。它们定义如何将分隔的文件读入行

      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 还支持以下不区分大小写的选项

属性描述
url要连接的JDBC URL。 源特定的连接属性可以在URL中指定。 例如jdbc:jdbc:postgresql://localhost/test?user=fred&password=secret
dbtable应该读取的 JDBC 表。请注意,可以使用在SQL查询的 FROM 子句中有效的任何内容。 例如,您可以使用括号中的子查询代替完整表
driver用于连接到此 URL 的 JDBC driver 程序的类名
partitionColumn, lowerBound, upperBound这三个属性必须被一起设置.partitionColumn 必须是数字列.lowerBound 和 upperBound 仅用于决定分区的大小而不是用于过滤表中的行.因此,表中的所有行将被分区并返回
numPartitions在表读写中可以用于并行度的最大分区数。这也确定并发JDBC连接的最大数量。 如果要写入的分区数超过此限制,则在写入之前通过调用 coalesce(numPartitions) 将其减少到此限制
fetchsizeJDBC 抓取的大小,用于确定每次数据往返传递的行数。 这有利于提升 JDBC driver 的性能,它们的默认值较小(例如: Oracle 是 10 行)。 该选项仅适用于读取操作
batchsizeJDBC 批处理的大小,用于确定每次数据往返传递的行数。 这有利于提升 JDBC driver 的性能。 该选项仅适用于写操作。默认值为 1000
isolationLevel

事务隔离级别,适用于当前连接。 它可以是 NONEREAD_COMMITTEDREAD_UNCOMMITTEDREPEATABLE_READ, 或 SERIALIZABLE 之一,对应于 JDBC 连接对象定义的标准事务隔离级别,默认为 READ_UNCOMMITTED

此选项仅适用于写操作。请参考 java.sql.Connection 中的文档

sessionInitStatement

在每个数据库会话打开到远程DB并开始读取数据之后,该选项将执行一个自定义SQL语句(或PL/SQL块)。使用它来实现会话初始化代码。

示例:选项(“sessionInitStatement”、“”开始执行即时的“alter session set”“_serial_direct_read”=true”;结束;”“”)

truncate这是一个与 JDBC 相关的选项.启用 SaveMode.Overwrite 时,此选项会导致 Spark 截断现有表,而不是删除并重新创建。它默认为 false。 此选项仅适用于写操作
createTableOptions这是一个与JDBC相关的选项。 如果指定,此选项允许在创建表时设置特定于数据库的表和分区选项(例如:CREATE TABLE t (name string) ENGINE=InnoDB. )。此选项仅适用于写操作
createTableColumnTypes这是一个与JDBC相关的选项.如果指定,此选项允许在创建表时设置特定于数据库的表和分区选项(例如:CREATE TABLE t (name string) ENGINE=InnoDB.).此选项仅适用于写操作
customSchema

用于从连接器读取JDBC数据的自定义模式。例如,“id DECIMAL(38,0), name STRING”。还可以指定部分字段,其他字段使用默认类型映射。

例如,“id DECIMAL(38,0)”。列名应该与JDBC表对应的列名相同。用户可以指定Spark SQL的相应数据类型,而不是使用默认值。此选项仅适用于读取

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

    原文作者:spark
    原文地址: https://www.cnblogs.com/NightPxy/p/9266125.html
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞