前言

自从前年(2020年)，我们团队将大数据平台升级到当时最新的HDP3.1(虽然当时HDP3.2也出来了，但是无法下载)，就遇到了这个坑，当时这个版本的spark2无法用传统的jdbc方式直接连接hive3，相信在2020年前后升级到最新版本HDP集群的小伙伴应该会遇到跟我一样的问题。

大数据生态的组件就是这样，本来之前在低版本时期大家都好好的，彼此之间[相濡以沫]，但是突然升级之后，反而[相看两厌]了，不兼容了起来。但好在，spark3之后，这一对冤家又重归于好。

老实说，当时为了解决这个不兼容问题，可算费了老劲了，但是当时集群都已经升级完成了，总不能因为这个不兼容的问题又给回退回去吧，那不是我的风格，于是，只能迎难而上，我当时的想法是，这么大的事，社区总不能不给解决方案吧，于是到官网的论坛逛了一圈发现，果然，Hortonworks给出了一个应急预案。

之所以想着写这篇文章，是想着既然我能遇到，屏幕前的你也很有可能遇到类似的问题，遂想着把过程和思考记录下来，与你分享，万一对你有用呢。

01

—

问题的暴露

当时我们直接通过spark2的api去读取hive3，当时的maven配置如下：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId><!--提供SparkContext对象-->
    <version>2.3.2</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.11</artifactId><!--提供SparkSession对象-->
    <version>2.3.2</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.11</artifactId><!--提供spark连接hive-->
    <version>2.3.2</version>
</dependency>

为了方便说明问题，用一段最简单读取hive表的代码来演示：

object Spark2ConnectHive3 {

    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local")
        conf.setAppName("connect_hive")
        val sparkSession = SparkSession.builder().config(conf)
                .enableHiveSupport()
                .getOrCreate()

        val result = sparkSession.sql("select * from doi_data limit 2")
        result.show()

        sparkSession.close()
        sparkSession.stop()
    }
}

按照往常的经验，运行代码，但是很不幸，运行出错了，报的错如下：

说Tez的session not running，很诡异，我不就是运行个select limit吗？怎么还需要Tez启动了吗？

那要不我提交到集群上试试(有时候任务提交到集群运行可以解决一些基础包依赖的问题)？

于是我满怀信心的提交，但是很快，运行结果就让我失望了：

反正这个错报的莫名其妙，我是没有看懂，于是，经过一番官网的查找发现了这么段话：

啥意思呢？HDP3.0之后spark(2.x)跟hive(3.x)的category独立管理，导致无法用之前的方案直接对hive进行访问，但是可以借助Hive Warehouse Connector这么个中间工具来实现交互。

也就是说之前低版本的方案是这样的：

spark api直接连接hive

而现在，对于spark2.x读取hive3.x，提供的解决方案是这样的：

spark api通过HWC连接hive

可以看到，在原来的基础上多了个HWC中间层，官方文档告诉我们，通过这种方式就可以解决刚才那个问题。

02

—

HWC解决方案

既然已经给了解决方案，那我们就用起来。

第一步：

根据官方的要求，需要打开hive的LLAP功能，我这里用的ambari，找到hive的配置界面，根据配置一步步操作(详细见官方文档)，如下图所示：

配置完成后，需要重启hive服务，重启后，会在yarn的任务管理界面多出几个跟llap相关的守护进程：

第二步：

说需要HWC，那么就在maven工程中把该包引入进来，加入对应的HWC依赖：

<dependency>
    <groupId>com.hortonworks</groupId><!--nexus私服自定义jar-->
    <artifactId>spark2-hive3-connector</artifactId>
    <version>1.0.0</version>
</dependency>

注意，因为目前官方只提供了离线下载这个jar包的方式，你直接通过这种方式引入是不行的，我这里是为了让项目组在开发时方便使用这个jar包，便把它加入到了本地的Nexus私服中。

第三步：

修改对应的API，如下所示：

object Spark2ConnectHive3 {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
        conf.setAppName("connect_hive")
        val sparkSession = SparkSession.builder().config(conf)
                .enableHiveSupport()
                .getOrCreate()

        //val result = sparkSession.sql("select * from doi_data limit 2")
        val hive = HiveWarehouseSession.session(sparkSession).build()//获取HWC对象
        val result = hive.executeQuery("select * from doi_data limit 2")//查询hive表数据
        result.show()

        sparkSession.close()
        sparkSession.stop()
    }
}

其配置信息hive-site.xml放在工程的resources目录下：

第四步：

提交运行，运行结果如下图所示：

为什么这里是提交到集群，因为在本地运行的话，会报一堆缺少依赖包而导致的各种报错，这也是这个版本的坑之一。

可以看到，现在已经能够正常跟hive进行通信了。

另外，在文档中其实还要求hive中的表必须满足ACID特性，才能用这种方式进行操作，只不过hive3的ACID特性默认开启，所谓的ACID属性，主要指下面几个配置：

hive.strict.managed.tables=true
hive.create.as.insert.only=true
metastore.create.as.acid=true

因为hive3默认已经打开，暂时不需要你做额外操作。

我看网上有文章说用非ACID的表也可以解决该问题，我亲自论证了一下，是不行的，如果不用HWC，即便你用到一种没有ACID属性的表，直接spark读取hive是行不通的，不信，你可以验证一下。

03

—

HWC方案的缺点

虽然提供的这个方案解决了spark2.x跟hive3.x的连通性问题，但是这个方案的缺点也比较明显。

必须启用llap功能，而llap必须要启动若干个守护进程，对集群资源有一定的侵占性；
需要额外引入HWC的jar包，用hwc提供的api对hive进行操作，但是经过我们一段时间的使用，不是特别好用，不时出现一些莫名其妙的bug，典型的如：无法在本地ieda环境直接运行代码，必须要提交到集群环境；还有就是通过该方式写数据到hive分区表的时候，明明已经制定好了分区字段，但是总能出现少许出乎意料的分区(已经确认不是数据问题)，如下图所示：

出现分区错乱

llap的守护进程是用来承载hive数据的写入和读取工作的，但是这个守护进程占用的资源必须提前配置，需要提前对后续数据的写入和读去提前预估，而这个预估其实很难，如果资源配置过高(CPU和内存)，会导致占用过多的yarn资源而造成浪费。如果配置过低，当集群中出现多个并行的spark操作hive的任务时，可能因为llap进程内存资源不足而导致OOM，而此类情况，我们已经遇到多次；
某些服务如果挂掉，比如hiveserver2 interactive，如果想重启，有时候特别困难，或者重启时间很长；
只支持ORC存储格式的表，且必须拥有ACID属性；

虽然该方案能解决spark2.x与hive3.x不兼容性问题，但是在平时开发、调试过程中总感觉没有之前低版本那样[丝滑]，每次修改完代码都必须得提交集群才能运行，不是很方便。

04

—

升级spark解决方案

无他，升级spark到3.x版本，因为当时就是因为spark的步子迈小了，没能跟上hive升级的步伐，需要升级到相同的版本，才能与hive平起平坐。

来，先升级maven依赖(此时就不需要HWC了)：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId><!--提供SparkContext对象-->
    <version>3.2.0</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId><!--提供SparkSession对象-->
    <version>3.2.0</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.12</artifactId><!--提供spark连接hive-->
    <version>3.2.0</version>
</dependency>

然后换成不用HWC的代码，先本地运行下试试：

object Spark3ConnectHive3 {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local")
        conf.setAppName("connect_hive")
        val sparkSession = SparkSession.builder().config(conf)
                .enableHiveSupport()
                .getOrCreate()

        val result = sparkSession.sql("select * from doi_data limit 2")
        result.show()

        sparkSession.close()
        sparkSession.stop()
    }
}

emm... 报错了：

报错原因说因为这是一张insert only表，经过一番查证发现，原来spark3如果不想通过HWC访问hive的表的话，就要求该表不能是ACID属性的表，但是因为hive3默认ACID属性开启。

我一直觉得hive的ACID属性是个鸡肋，这个像是让其功能更加贴近RDBMS而强加进去的，除了能让你对hive表进行update、细粒度的insert之外(效率很低)，并不能实现像传统数据库那样的事务功能。

因此，只有新建一张非ACID属性的表，如下建表语句：

CREATE EXTERNAL TABLE `as_bgp_bak`(               
   `as_start` string,                             
   `as_end` string,                               
   `id` int)                                      
 ROW FORMAT SERDE                                 
   'org.apache.hadoop.hive.ql.io.orc.OrcSerde'    
 STORED AS INPUTFORMAT                            
   'org.apache.hadoop.hive.ql.io.orc.OrcInputFormat' 
 OUTPUTFORMAT                                       
   'org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat'
 TBLPROPERTIES (                                    
  'transactional'='false'             
)

（请原谅我用了一张新的表面，原来那张被我误删了），然后用insert overwrite语句写入些数据进去；

然后就直接在本地运行该代码(居然可以在本地运行，之前spark2是不可以的)，运行结果如下：

可能有同学有疑问，说spark升级难道不需要升级spark服务吗？我的经验是：不需要，因为它不像hive、HDFS这一类[强服务]的组件，spark是[弱服务]的，升级对应的jar包即可，具体如何升级，后面文章会具体说明。

（至于升级spark之后这个方案效果如何，还需等待一段时间验证后再来告诉你）

05

—

升级过程的注意事项

在升级过程中，如果是采用HWC解决方案的话，过程会很曲折，官方虽然提供了基于HWC的api，但是当你真正在使用的时候你会发现，很多地方不一定好使，需要你不断的去试错，很难一次就成功，所以一定要有耐心。

而如果你是升级spark的话(从spark2.x到spark3.x)，需要注意的是spark2支持的Scala版本最低为2.11，而spark3支持的Scala版本最低为2.12，如果你跟我一样，在使用spark2.x的时候刚好Scala版本为2.11，那么将spark升级到spark3.x后，此时一定要记得将Scala也一并升级过来(最低要求2.12)，否则运行会报一些如下图所示，Scala的基础类无法找到的错。

结语

——

在大数据集群的开发和使用过程中，类似这种问题总是少不了，如果你是个大数据开发老手，相信也一定遇到过不少类似[蛋疼]的问题，谨以此篇文章，希望能够与你产生些许共鸣。

因公众号无法留言，如果你觉得我讲的还凑合或者有任何疑问，欢迎找我交流探讨...

ITPub博客

spark2 + hive3遇到的坑

最新文章