spark系列18：sparksql操作hive表的使用

桃扇骨 2022-11-10 14:27 271阅读 0赞

#  1.SparkSQL 整合 Hive #

导读

1.  开启 `Hive` 的 `MetaStore` 独立进程
2.  整合 `SparkSQL` 和 `Hive` 的 `MetaStore`

和一个文件格式不同, `Hive` 是一个外部的数据存储和查询引擎, 所以如果 `Spark` 要访问 `Hive` 的话, 就需要先整合 `Hive`

整合什么 ?

如果要讨论 `SparkSQL` 如何和 `Hive` 进行整合, 首要考虑的事应该是 `Hive` 有什么, 有什么就整合什么就可以

*  `MetaStore`, 元数据存储
    
    `SparkSQL` 内置的有一个 `MetaStore`, 通过嵌入式数据库 `Derby` 保存元信息, 但是对于生产环境来说, 还是应该使用 `Hive` 的 `MetaStore`, 一是更成熟, 功能更强, 二是可以使用 `Hive` 的元信息
 *  查询引擎
    
    `SparkSQL` 内置了 `HiveSQL` 的支持, 所以无需整合

为什么要开启 `Hive` 的 `MetaStore`

`Hive` 的 `MetaStore` 是一个 `Hive` 的组件, 一个 `Hive` 提供的程序, 用以保存和访问表的元数据, 整个 `Hive` 的结构大致如下

![20190523011946][]

由上图可知道, 其实 `Hive` 中主要的组件就三个, `HiveServer2` 负责接受外部系统的查询请求, 例如 `JDBC`, `HiveServer2` 接收到查询请求后, 交给 `Driver` 处理, `Driver` 会首先去询问 `MetaStore` 表在哪存, 后 `Driver` 程序通过 `MR` 程序来访问 `HDFS` 从而获取结果返回给查询请求者

而 `Hive` 的 `MetaStore` 对 `SparkSQL` 的意义非常重大, 如果 `SparkSQL` 可以直接访问 `Hive` 的 `MetaStore`, 则理论上可以做到和 `Hive` 一样的事情, 例如通过 `Hive` 表查询数据

而 Hive 的 MetaStore 的运行模式有三种

*  内嵌 `Derby` 数据库模式
    
    这种模式不必说了, 自然是在测试的时候使用, 生产环境不太可能使用嵌入式数据库, 一是不稳定, 二是这个 `Derby` 是单连接的, 不支持并发
 *  `Local` 模式
    
    `Local` 和 `Remote` 都是访问 `MySQL` 数据库作为存储元数据的地方, 但是 `Local` 模式的 `MetaStore` 没有独立进程, 依附于 `HiveServer2` 的进程
 *  `Remote` 模式
    
    和 `Loca` 模式一样, 访问 `MySQL` 数据库存放元数据, 但是 `Remote` 的 `MetaStore` 运行在独立的进程中

我们显然要选择 `Remote` 模式, 因为要让其独立运行, 这样才能让 `SparkSQL` 一直可以访问

**`Hive` 开启 `MetaStore`**

`Step 1`: 修改 `hive-site.xml`

<property>
      <name>hive.metastore.warehouse.dir</name>
      <value>/user/hive/warehouse</value>
    </property>
    
    <property>
      <name>javax.jdo.option.ConnectionURL</name>
      <value>jdbc:mysql://node01:3306/hive?createDatabaseIfNotExist=true</value>
    </property>
    
    <property>
      <name>javax.jdo.option.ConnectionDriverName</name>
      <value>com.mysql.jdbc.Driver</value>
    </property>
    
    <property>
      <name>javax.jdo.option.ConnectionUserName</name>
      <value>username</value>
    </property>
    
    <property>
      <name>javax.jdo.option.ConnectionPassword</name>
      <value>password</value>
    </property>
    
    <property>
      <name>hive.metastore.local</name>
      <value>false</value>
    </property>
    
    <property>
      <name>hive.metastore.uris</name>
      <value>thrift://node01:9083</value>  //当前服务器
    </property>

**`Step 2`: 启动 `Hive MetaStore`**

nohup /export/servers/hive/bin/hive --service metastore 2>&1 >> /var/log.log &

`SparkSQL` 整合 `Hive` 的 `MetaStore`

即使不去整合 `MetaStore`, `Spark` 也有一个内置的 `MateStore`, 使用 `Derby` 嵌入式数据库保存数据, 但是这种方式不适合生产环境, 因为这种模式同一时间只能有一个 `SparkSession` 使用, 所以生产环境更推荐使用 `Hive` 的 `MetaStore`

`SparkSQL` 整合 `Hive` 的 `MetaStore` 主要思路就是要通过配置能够访问它, 并且能够使用 `HDFS` 保存 `WareHouse`, 这些配置信息一般存在于 `Hadoop` 和 `HDFS` 的配置文件中, 所以可以直接拷贝 `Hadoop` 和 `Hive` 的配置文件到 `Spark` 的配置目录

cd /export/servers/hadoop/etc/hadoop
    cp hive-site.xml core-site.xml hdfs-site.xml /export/servers/spark/conf/   
    
    scp -r /export/servers/spark/conf node02:/export/servers/spark/conf
    scp -r /export/servers/spark/conf node03:/export/servers/spark/conf

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td><code>Spark</code>&nbsp;需要&nbsp;<code>hive-site.xml</code>&nbsp;的原因是, 要读取&nbsp;<code>Hive</code>&nbsp;的配置信息, 主要是元数据仓库的位置等信息</td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
   <td><code>Spark</code>&nbsp;需要&nbsp;<code>core-site.xml</code>&nbsp;的原因是, 要读取安全有关的配置</td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
   <td><code>Spark</code>&nbsp;需要&nbsp;<code>hdfs-site.xml</code>&nbsp;的原因是, 有可能需要在&nbsp;<code>HDFS</code>&nbsp;中放置表文件, 所以需要&nbsp;<code>HDFS</code>&nbsp;的配置</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>如果不希望通过拷贝文件的方式整合 Hive, 也可以在 SparkSession 启动的时候, 通过指定 Hive 的 MetaStore 的位置来访问, 但是更推荐整合的方式</p> </td> 
  </tr> 
 </tbody> 
</table>

# 2. 访问 Hive 表 #

导读

1.  在 `Hive` 中创建表
2.  使用 `SparkSQL` 访问 `Hive` 中已经存在的表
3.  使用 `SparkSQL` 创建 `Hive` 表
4.  使用 `SparkSQL` 修改 `Hive` 表中的数据

在 `Hive` 中创建表

第一步, 需要先将文件上传到集群中, 使用如下命令上传到 `HDFS` 中

hdfs dfs -mkdir -p /dataset
    hdfs dfs -put studenttabl10k /dataset/

第二步, 使用 `Hive` 或者 `Beeline` 执行如下 `SQL`

CREATE DATABASE IF NOT EXISTS spark_integrition;
    
    USE spark_integrition;
    
    CREATE EXTERNAL TABLE student
    (
      name  STRING,
      age   INT,
      gpa   string
    )
    ROW FORMAT DELIMITED
      FIELDS TERMINATED BY '\t'
      LINES TERMINATED BY '\n'
    STORED AS TEXTFILE
    LOCATION '/dataset/hive';
    
    LOAD DATA INPATH '/dataset/studenttab10k' OVERWRITE INTO TABLE student;

通过 `SparkSQL` 查询 `Hive` 的表

查询 `Hive` 中的表可以直接通过 `spark.sql(…)` 来进行, 可以直接在其中访问 `Hive` 的 `MetaStore`, 前提是一定要将 `Hive` 的配置文件拷贝到 `Spark` 的 `conf` 目录

scala> spark.sql("use spark_integrition")
    scala> val resultDF = spark.sql("select * from student limit 10")
    scala> resultDF.show()

通过 `SparkSQL` 创建 `Hive` 表

通过 `SparkSQL` 可以直接创建 `Hive` 表, 并且使用 `LOAD DATA` 加载数据

val createTableStr =
      """
        |CREATE EXTERNAL TABLE student
        |(
        |  name  STRING,
        |  age   INT,
        |  gpa   string
        |)
        |ROW FORMAT DELIMITED
        |  FIELDS TERMINATED BY '\t'
        |  LINES TERMINATED BY '\n'
        |STORED AS TEXTFILE
        |LOCATION '/dataset/hive'
      """.stripMargin
    
    spark.sql("CREATE DATABASE IF NOT EXISTS spark_integrition1")
    spark.sql("USE spark_integrition1")
    spark.sql(createTableStr)
    spark.sql("LOAD DATA INPATH '/dataset/studenttab10k' OVERWRITE INTO TABLE student")
    spark.sql("select * from student limit").show()

目前 `SparkSQL` 支持的文件格式有 `sequencefile`, `rcfile`, `orc`, `parquet`, `textfile`, `avro`, 并且也可以指定 `serde` 的名称

使用 `SparkSQL` 处理数据并保存进 Hive 表

前面都在使用 `SparkShell` 的方式来访问 `Hive`, 编写 `SQL`, 通过 `Spark` 独立应用的形式也可以做到同样的事, 但是需要一些前置的步骤, 如下

Step 1: 导入 `Maven` 依赖

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

Step 2: 配置 `SparkSession`

如果希望使用 `SparkSQL` 访问 `Hive` 的话, 需要做两件事

1.  开启 `SparkSession` 的 `Hive` 支持
    
    经过这一步配置, `SparkSQL` 才会把 `SQL` 语句当作 `HiveSQL` 来进行解析
2.  设置 `WareHouse` 的位置
    
    虽然 `hive-stie.xml` 中已经配置了 `WareHouse` 的位置, 但是在 `Spark 2.0.0` 后已经废弃了 `hive-site.xml` 中设置的 `hive.metastore.warehouse.dir`, 需要在 `SparkSession` 中设置 `WareHouse` 的位置
3.  设置 `MetaStore` 的位置

val spark = SparkSession
      .builder()
      .appName("hive example")
      .config("spark.sql.warehouse.dir", "hdfs://node01:8020/dataset/hive")  
      .config("hive.metastore.uris", "thrift://node01:9083")                 
      .enableHiveSupport()                                                   
      .getOrCreate()

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>设置&nbsp;<code>WareHouse</code>&nbsp;的位置</td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
   <td>设置&nbsp;<code>MetaStore</code>&nbsp;的位置</td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
   <td>开启&nbsp;<code>Hive</code>&nbsp;支持</td> 
  </tr> 
 </tbody> 
</table>

配置好了以后, 就可以通过 `DataFrame` 处理数据, 后将数据结果推入 `Hive` 表中了, 在将结果保存到 `Hive` 表的时候, 可以指定保存模式

val schema = StructType(
      List(
        StructField("name", StringType),
        StructField("age", IntegerType),
        StructField("gpa", FloatType)
      )
    )
    
    val studentDF = spark.read
      .option("delimiter", "\t")
      .schema(schema)
      .csv("dataset/studenttab10k")
    
    val resultDF = studentDF.where("age < 50")
    
    resultDF.write.mode(SaveMode.Overwrite).saveAsTable("spark_integrition1.student")

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>通过&nbsp;<code>mode</code>&nbsp;指定保存模式, 通过&nbsp;<code>saveAsTable</code>&nbsp;保存数据到&nbsp;<code>Hive</code></td> 
  </tr> 
 </tbody> 
</table>

[20190523011946]: /images/20221022/4bbd67267fc248618e2b93bec8fd25ba.png