Spark系列--SparkSQL(六)数据源

梦里梦外; 2022-05-19 10:13 255阅读 0赞

## 前言 ##

SparkSQL的数据源：结构化的文件（json，parquet），或者是Hive的表，或者是外部的数据库（mysql），也或者是已经存在的RDD。

Spark SQL的DataFrame接口支持多种数据源的操作。一个DataFrame可以进行RDDs方式的操作，也可以被注册为临时表。把DataFrame注册为临时表之后，就可以对该DataFrame执行SQL查询。

Spark SQL的默认数据源为Parquet格式。数据源为Parquet文件时，Spark SQL可以方便的执行所有的操作。修改配置项spark.sql.sources.default，可修改默认数据源格式。

## 一、通用加载/保存方法 ##

### 1.1 手动指定选项 ###

*  Load是用来读取文件的时候加载文件中的数据
 *  Save是用来往外写文件的时候存储写出的数据

val df = spark.read.load("examples/src/main/resources/users.parquet") 
    
    df.select("name", "favorite_color").write.save("namesAndFavColors.parquet")

当数据源格式不是parquet格式文件时，需要手动指定数据源的格式。数据源格式需要指定全名（例如：org.apache.spark.sql.parquet），如果数据源格式为内置格式，则只需要指定简称定json, parquet, jdbc, orc, libsvm, csv, text来指定数据的格式。  
可以通过SparkSession提供的read.load方法用于通用加载数据，使用write和save保存数据。

val peopleDF = spark.read.format("json").load("examples/src/main/resources/people.json") 
    
    peopleDF.write.format("parquet").save("hdfs://master01:9000/namesAndAges.parquet")

除此之外，可以直接运行SQL在文件上：

val sqlDF = spark.sql("SELECT * FROM parquet.`hdfs://hadoop0:9000/namesAndAges.parquet`") sqlDF.show()

### 1.2 文件保存选项 ###

可以采用SaveMode执行存储操作，SaveMode定义了对数据的处理模式。需要注意的是，这些保存模式不使用任何锁定，不是原子操作。此外，当使用Overwrite方式执行时，在输出新数据之前原数据就已经被删除。SaveMode详细介绍如下表：

<table> 
 <thead> 
  <tr> 
   <th>Scala/Java</th> 
   <th align="center">Any Language</th> 
   <th align="right">Meaning</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>SaveMode.ErrorIfExists(default)</td> 
   <td align="center">“error”(default)</td> 
   <td align="right">如果文件存在，则报错</td> 
  </tr> 
  <tr> 
   <td>SaveMode.Append</td> 
   <td align="center">“append”</td> 
   <td align="right">追加</td> 
  </tr> 
  <tr> 
   <td>SaveMode.Overwrite</td> 
   <td align="center">“overwrite”</td> 
   <td align="right">覆写</td> 
  </tr> 
  <tr> 
   <td>SaveMode.Ignore</td> 
   <td align="center">“ignore”</td> 
   <td align="right">数据存在，则忽略</td> 
  </tr> 
 </tbody> 
</table>

--------------------

## 二、Parquet文件 ##

Parquet是一种流行的列式存储格式，可以高效地存储具有嵌套字段的记录。

![这里写图片描述][70]

### 2.1 Parquet读写 ###

Parquet格式经常在Hadoop生态圈中被使用，它也支持Spark SQL的全部数据类型。Spark SQL 提供了直接读取和存储 Parquet 格式文件的方法。

import spark.implicits._
    
    val peopleDF = spark.read.json("examples/src/main/resources/people.json")  
    
    peopleDF.write.parquet("hdfs://hadoop0:9000/people.parquet") 
    
    val parquetFileDF = spark.read.parquet("hdfs://hadoop0:9000/people.parquet") 
    
    parquetFileDF.createOrReplaceTempView("parquetFile") 
    
    val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")
    
     namesDF.map(attributes => "Name: " + attributes(0)).show()

### 2.2 解析分区信息 ###

对表进行分区是对数据进行优化的方式之一。在分区的表内，数据通过分区列将数据存储在不同的目录下。Parquet数据源现在能够自动发现并解析分区信息。例如，对人口数据进行分区存储，分区列为gender和country，使用下面的目录结构：

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给 SQLContext.read.parquet或SQLContext.read.load，Spark SQL将自动解析分区信息。返回的DataFrame的Schema如下：

root
    |-- name: string (nullable = true)
    |-- age: long (nullable = true)
    |-- gender: string (nullable = true)
    |-- country: string (nullable = true)

需要注意的是，数据的分区列的数据类型是自动解析的。当前，支持数值类型和字符串类型。自动解析分区类型的参数为：spark.sql.sources.partitionColumnTypeInference.enabled，默认值为true。如果想关闭该功能，直接将该参数设置为disabled。此时，分区列数据格式将被默认设置为string类型，不再进行类型解析。

### 2.3 Schema合并 ###

像ProtocolBuffer、Avro和Thrift那样，Parquet也支持Schema evolution（Schema演变）。用户可以先定义一个简单的Schema，然后逐渐的向Schema中增加列描述。通过这种方式，用户可以获取多个有不同Schema但相互兼容的Parquet文件。现在Parquet数据源能自动检测这种情况，并合并这些文件的schemas。

因为Schema合并是一个高消耗的操作，在大多数情况下并不需要，所以Spark SQL从1.5.0开始默认关闭了该功能。可以通过下面两种方式开启该功能：

*  当数据源为Parquet文件时，将数据源选项mergeSchema设置为true
 *  设置全局SQL选项spark.sql.parquet.mergeSchema为true

import spark.implicits._
    
    val df1 = sc.makeRDD(1 to 5).map(i => (i, i * 2)).toDF("single", "double") 
    
    df1.write.parquet("hdfs://hadoop0:9000/data/test_table/key=1") 
    
    val df2 = sc.makeRDD(6 to 10).map(i => (i, i * 3)).toDF("single", "triple") 
    
    df2.write.parquet("hdfs://hadoop0:9000/data/test_table/key=2") 
    
    val df3 = spark.read.option("mergeSchema", "true").parquet("hdfs://master01:9000/data/test_table")
    
     df3.printSchema()

结果如下：

root
    |-- single: int (nullable = true) 
    |-- double: int (nullable = true) 
    |-- triple: int (nullable = true) 
    |-- key : int (nullable = true)

### 2.4 配置 ###

parquet的配置可以通过SQLContext的setConf方法或执行set key=balue命令行：

## <table>   <thead>    <tr>     <th>Property Name</th>     <th align="center">Default</th>     <th align="right">Meaning</th>    </tr>   </thead>   <tbody>    <tr>     <td>spark.sql.parquet.binaryAsString</td>     <td align="center">false</td>     <td align="right">一些其他Parquet-producing systems, 例如 Impala, Hive, 旧版本的 Spark SQL, 在导出parquet schema时不区分二进制数据和string 这个选项告诉Spark SQL把二进制数据当作string来保证兼容性.</td>    </tr>    <tr>     <td>spark.sql.parquet.int96AsTimestamp</td>     <td align="center">true</td>     <td align="right">一些 Parquet-producing systems,例如 Impala 和Hive, 存储 Timestamp into INT96. 这个选项告诉Spark SQL把INT96做为timestamp来保证兼容性.</td>    </tr>    <tr>     <td>spark.sql.parquet.cacheMetadata</td>     <td align="center">true</td>     <td align="right">开启能缓存parquet schema metadata,当查询静态数据时能提升效率.</td>    </tr>    <tr>     <td>spark.sql.parquet.compression.codec</td>     <td align="center">gzip</td>     <td align="right">设置写parquet文件时的压缩码率，可选的值：uncompressed, snappy, gzip, lzo.</td>    </tr>    <tr>     <td>spark.sql.parquet.filterPushdown</td>     <td align="center">true</td>     <td align="right">当为true能优化parquet filter push-down</td>    </tr>    <tr>     <td>spark.sql.hive.convertMetastoreParquet</td>     <td align="center">true</td>     <td align="right">当为 false, Spark SQL 会使用 Hive SerDe 解析 parquet tables 而不是默认spark提供的</td>    </tr>    <tr>     <td>spark.sql.parquet.output.committer.class</td>     <td align="center">org.apache.parquet.hadoop.ParquetOutputCommitter</td>     <td align="right">parquet用到的输出类 . 指定的类需要是 org.apache.hadoop.mapreduce.OutputCommitter的子类. 通常,它也是org.apache.parquet.hadoop.ParquetOutputCommitter的子类.Note:这个选项会被忽视如果 spark.speculation 为开启的.这个选项必须通过 Hadoop Configuration 来配置而不是 Spark SQLConf.这个选项重写了spark.sql.sources.outputCommitterClass.Spark SQL 有一个内置的类org.apache.spark.sql.parquet.DirectParquetOutputCommitter, 可以更高效的写入paruquet到S3</td>    </tr>    <tr>     <td>spark.sql.parquet.mergeSchema</td>     <td align="center">false</td>     <td align="right">当为 true, parquet数据源从所有数据文件合并schema , 不然schema要从合并文件或随机文件里选择(若没有合并文件).</td>    </tr>   </tbody>  </table> ##

## 三、Hive数据库 ##

Apache Hive是Hadoop上的SQL引擎，Spark SQL编译时可以包含Hive支持，也可以不包含。包含Hive支持的Spark SQL可以支持Hive表访问、UDF(用户自定义函数)以及 Hive 查询语言(HiveQL/HQL)等。需要强调的一点是，如果要在Spark SQL中包含Hive的库，并不需要事先安装Hive。一般来说，最好还是在编译Spark SQL时引入Hive支持，这样就可以使用这些特性了。**如果你下载的是二进制版本的 Spark，它应该已经在编译时添加了 Hive 支持。**

若要把Spark SQL连接到一个部署好的Hive上，你必须把hive-site.xml复制到 Spark的配置文件目录中($SPARK\_HOME/conf)。即使没有部署好Hive，Spark SQL也可以运行。 需要注意的是，如果你没有部署好Hive，Spark SQL会在当前的工作目录中创建出自己的Hive 元数据仓库，叫作 metastore\_db。此外，如果你尝试使用 HiveQL 中的 CREATE TABLE (并非 CREATE EXTERNAL TABLE)语句来创建表，这些表会被放在你默认的文件系统中的 /user/hive/warehouse 目录中(如果你的 classpath 中有配好的 hdfs-site.xml，默认的文件系统就是 HDFS，否则就是本地文件系统)。

[Spark系列(十五)SparkSQL–执行SparkSQL查询][Spark_SparkSQL_SparkSQL]  
IDEA使用Spark内嵌Hive应用，在原先SparkSQL项目依赖基础上增加Hive依赖

<dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>${spark.version}</version>
            
        </dependency>

**指定spark.sql.warehouse.dir为本地绝对路径，如果为HDFS路径，请注意看后面内嵌Hive应用注意事项**。

**完整代码如下：**

package com.m.jd
    
    import java.io.File
    
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.{DataFrame, SparkSession}
    
    object HiveOperation extends App { 
    
      val warehouseLocation = new File("/opt/spark-warehouse").getAbsolutePath
    
      private val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark Hive Example")
    
      private val session: SparkSession =   SparkSession.builder().appName("Spark Hive Example").config(sparkConf).config("spark.sql.warehouse.dir",warehouseLocation).enableHiveSupport().getOrCreate()
    
      import session.implicits._
      import org.apache.spark.sql._
      import session.sql
    
      sql("CREATE TABLE IF NOT EXISTS test (key INT, value STRING) row format delimited fields terminated by '\t'")
    
      sql("LOAD DATA LOCAL INPATH '/opt/spark-data/kv1.txt' INTO TABLE test")
    
      private val df: DataFrame = sql("SELECT * FROM test")
    
      df.show()
    
    +---+------+
    |key| value|
    +---+------+
    | 88|val_88|
    | 67|val_67|
    +---+------+
    
      val sqlDF = sql("SELECT key, value FROM test WHERE key > 10 ORDER BY key")
    
      private val stringsDS: Dataset[String] = sqlDF.map{
       case Row(key: Int, value: String) => s"Key-value:$key-$value"}
    
      stringsDS.show()
    
    +-------------------+
    |              value|
    +-------------------+
    |Key-value:67-val_67|
    |Key-value:88-val_88|
    +-------------------+
    
      session.close()
    }

### 3.1 内嵌Hive应用 ###

如果要使用内嵌的Hive，什么都不用做，直接用就可以了。

--conf : spark.sql.warehouse.dir=

![这里写图片描述][70 1]

注意：如果你使用的是内部的Hive，在Spark2.0之后，spark.sql.warehouse.dir用于指定数据仓库的地址，如果你需要是用HDFS作为路径，那么需要将core-site.xml和hdfs-site.xml 加入到Spark conf目录，否则只会创建master节点上的warehouse目录，查询时会出现文件找不到的问题，这时需要使用HDFS，则需要将metastore删除，重启集群。

### 3.2 外部Hive应用 ###

如果想连接外部已经部署好的Hive，需要通过以下几个步骤。

**1）将Hive中的hive-site.xml拷贝或者软连接到Spark安装目录下的conf目录下。**  
**2）打开spark shell，注意带上访问Hive元数据库的JDBC客户端。**

$ bin/spark-shell --master spark://hadoop0:7077 --jars mysql-connector-java-5.1.27-bin.jar

**注意，需要将相关的数据库驱动放到spark的类路径下($SPARK\_HOME/jars)。**

--------------------

## 四、JSON数据集 ##

Spark SQL 能够自动推测 JSON数据集的结构，并将它加载为一个Dataset\[Row\]. 可以通过SparkSession.read.json()去加载一个 Dataset\[String\]或者一个JSON 文件.注意，这个JSON文件不是一个传统的JSON文件，每一行都得是一个JSON串。

前面已经介绍了很多SparkSession.read.json()去加载一个JSON文件的例子了，这里介绍去加载一个Dataset\[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| 
    // +---------------+----+

## 五、JDBC ##

Spark SQL可以通过JDBC从关系型数据库中读取数据的方式创建DataFrame，通过对DataFrame一系列的计算后，还可以将数据再写回关系型数据库中。

**注意，需要将相关的数据库驱动放到spark的类路径下。**

$ bin/spark-shell --master spark://hadoop0:7077 --jars mysql-connector-java-5.1.27-bin.jar

// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
    // Loading data from a JDBC source
    
    val jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://hadoop0:3306/rdd").option("dbtable", " rddtable").option("user", "root").option("password", "123456").load()
    
    val connectionProperties = new Properties()
    connectionProperties.put("user", "root")
    connectionProperties.put("password", "123456")
    
    val jdbcDF2 = spark.read
    .jdbc("jdbc:mysql://hadoop0:3306/rdd", "rddtable", connectionProperties)
    
    // Saving data to a JDBC source
    
    jdbcDF.write
    .format("jdbc")
    .option("url", "jdbc:mysql://hadoop0:3306/rdd")
    .option("dbtable", "rddtable2")
    .option("user", "root")
    .option("password", "123456")
    .save()
    
    jdbcDF2.write
    .jdbc("jdbc:mysql://master01:3306/mysql", "db", connectionProperties)
    
    // Specifying create table column data types on write
    
    jdbcDF.write
    .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")
    .jdbc("jdbc:mysql://master01:3306/mysql", "db", connectionProperties)

参考文章：[https://blog.csdn.net/wsdc0521/article/details/50011349][https_blog.csdn.net_wsdc0521_article_details_50011349]

[70]: /images/20220519/c3e01c312113496c89c2a919925930d5.png
[Spark_SparkSQL_SparkSQL]: https://blog.csdn.net/u012834750/article/details/81164990
[70 1]: /images/20220519/6fd3d1595f9a462bbcd83b6db7a73d6f.png
[https_blog.csdn.net_wsdc0521_article_details_50011349]: https://blog.csdn.net/wsdc0521/article/details/50011349