我们经常会选择使用spark做原始数据的ETL清洗，清洗后的Dataframe最终需要写入到数仓的Hive表中，也就是说spark能否直接读写hive表特别重要。

spark on hive原理

要让spark程序可以直接读写hive表，只需要让spark可以访问到hive表的meta信息即可，因为spark自己实现了一套和hive一样的SQL引擎并且底层直接用spark运算，其SQL已经支持了hive的大部分特性。

下面是Hive和Spark的关系：

说白了，spark把hive又实现了一遍，为了和hive可以联动所以支持直接操作hive的metastore来共享元数据。

这里容易和hive on spark混淆，hive on spark是将hive的计算引擎换成spark，目的是为了加速计算。而这里的spark on hive是为了可以让spark可以读写hive表，以便可以通过Java编程形式的方式实现灵活的数据计算。

配置spark on hive很简单，把hive-site.xml软链接到spark的conf目录下即可，spark会自动读取该配置来识别metastore的地址。

ln -s /root/bigdata/hive-3.1.3/conf/hive-site.xml /root/bigdata/spark-3.0.1/conf/hive-site.xml

这就配置完成了，下面我们编写一个具体的spark程序作为示例，项目地址：https://github.com/owenliang/spark-sql-demo。

引入maven依赖

在《spark系列 – spark-sql编程使用》中，我演示了如何通过spark-sql库操作Dataframe以及直接写Hive格式的数据到HDFS指定路径下。

但是我们要注意，spark-sql虽然可以将DataFrame输出到HDFS上作为Hive表的数据，但是并没有修改过metastore中的元数据，所以并没有真的与HIve联动，Hive侧是看不到对应的表或者数据的。

要让spark-sql库可以直接读写metastore，我们需要引入另外spark-hive库来支持hive：

    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.0.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.3.0</version>
        </dependency>
        <dependency> <!-- Spark dependency -->
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.0.1</version>
            <scope>provided</scope>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-hive -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>3.0.1</version>
        </dependency>
    </dependencies>

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.12</artifactId>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-client</artifactId>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-sql_2.12</artifactId>

<scope>provided</scope>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-hive_2.12</artifactId>

</dependency>

</dependencies>

编写spark操作hive

接下来我们用spark直接创建hive数据库和表结构，然后将DataFrame的数据插入到hive表里。

初始化SparkSession需要额外调用enableHiveSupport()开启Hive支持，它会去spark的conf目录读取hive-site.xml配置获取metastore地址，其他代码没有变化：

public class Main {
    public static void main(String []args) {
        try {
            // spark sql session
            SparkSession sess = SparkSession.builder().appName("spark-sql-demo").enableHiveSupport().getOrCreate();
            // hdfs
            FileSystem dfs =  FileSystem.get(new Configuration());

public class Main {

public static void main(String []args) {

try {

// spark sql session

SparkSession sess = SparkSession.builder().appName("spark-sql-demo").enableHiveSupport().getOrCreate();

// hdfs

FileSystem dfs = FileSystem.get(new Configuration());

既然spark已经和hive直通了，我们可以像使用hive cli一样直接写Hive sql：

package cc.yuerblog.sql;

import org.apache.hadoop.fs.FileSystem;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;

public class HiveTest {
    public void run(FileSystem dfs, SparkSession sess) throws IOException {
        // 建hive库
        sess.sql("create database if not exists spark  ");
        // 切库
        sess.sql("use spark");
        // 建表
        sess.sql("create table if not exists people(name string, age int) stored as orc");

        // 加载JSON DataFrame
        Dataset<Row> df = sess.read().json("/people.txt");
        // 查看DataFrame
        df.show();
        // 创建视图
        df.createOrReplaceTempView("people_view");
        // 追加到表里（注意通过select调整视图表的字段顺序与hive表一致）
        sess.sql("insert into table people select name,age from people_view");
    }
}

package cc.yuerblog.sql;

import org.apache.hadoop.fs.FileSystem;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

import java.io.IOException;

public class HiveTest {

public void run(FileSystem dfs, SparkSession sess) throws IOException {

// 建hive库

sess.sql("create database if not exists spark ");

// 切库

sess.sql("use spark");

// 建表

sess.sql("create table if not exists people(name string, age int) stored as orc");

// 加载JSON DataFrame

Dataset<Row> df = sess.read().json("/people.txt");

// 查看DataFrame

df.show();

// 创建视图

df.createOrReplaceTempView("people_view");

// 追加到表里（注意通过select调整视图表的字段顺序与hive表一致）

sess.sql("insert into table people select name,age from people_view");

}

创建数据库：create database if not exists spark
使用数据库：use spark
创建people表：create table if not exists people(name string, age int) stored as orc，采用Orc列存储格式，Hive表一般采用列存来提升查询效率，压缩数据量。
加载DataFrame：通过sparkSession加载json文件/people.txt，里面每一行是json字符串，包含name和age两个字段。
创建视图：把Dataframe创建为视图表people_view，它只是逻辑视图，实际做计算时才会通过DAG图计算出来，不会真的存在Hive里。
插入people表：把视图表的数据select出来，插入到hive people表中去。（这里需要注意select字段的数量和顺序必须和hive表结构完全一致，hive插入时就是这样要求的，并不是spark特殊）

运行spark程序

将上述代码打包成jar，然后提交：

spark-submit –master yarn –deploy-mode client –executor-memory 1G –num-executors 1 ./spark-sql-demo-1.0-SNAPSHOT.jar

查看df.show()打印了DataFrame表格：

2020-10-14 05:53:02,225 INFO codegen.CodeGenerator: Code generated in 13.879469 ms
+---+---------+
|age|     name|
+---+---------+
| 83| person-0|
| 96| person-1|
| 96| person-2|
| 34| person-3|
| 93| person-4|
| 64| person-5|
| 25| person-6|
| 33| person-7|
| 91| person-8|
| 72| person-9|
| 90|person-10|
| 74|person-11|
| 68|person-12|
| 66|person-13|
| 93|person-14|
| 92|person-15|
| 79|person-16|
| 36|person-17|
| 84|person-18|
| 51|person-19|

2020-10-14 05:53:02,225 INFO codegen.CodeGenerator: Code generated in 13.879469 ms

+---+---------+

|age| name|

+---+---------+

| 83| person-0|

| 96| person-1|

| 96| person-2|

| 34| person-3|

| 93| person-4|

| 64| person-5|

| 25| person-6|

| 33| person-7|

| 91| person-8|

| 72| person-9|

| 90|person-10|

| 74|person-11|

| 68|person-12|

| 66|person-13|

| 93|person-14|

| 92|person-15|

| 79|person-16|

| 36|person-17|

| 84|person-18|

| 51|person-19|

利用hive cli查看hive表：

hive> select * from spark.people limit 10;
OK
people.name people.age
person-0 83
person-1 96
person-2 96
person-3 34
person-4 93
person-5 64
person-6 25
person-7 33
person-8 91
person-9 72
Time taken: 2.554 seconds, Fetched: 10 row(s)

本篇博客就这些内容，spark操作hive体验和hive基本一致，但是可以把dataframe转成view并通过hive sql直接访问是很实用的特性。

如果文章帮助您解决了工作难题，您可以帮我点击屏幕上的任意广告，或者赞助少量费用来支持我的持续创作，谢谢~

鱼儿的博客

hive系列 – spark on hive配置与编程

spark on hive原理

引入maven依赖

编写spark操作hive

运行spark程序

《hive系列 – spark on hive配置与编程》有1个想法

发表评论取消回复

spark on hive原理

引入maven依赖

编写spark操作hive

运行spark程序

《hive系列 – spark on hive配置与编程》有1个想法

发表评论 取消回复

发表评论取消回复