大数据开发之 Hive

​

Hive 简介

定义

Facebook 为了解决海量日志数据的分析而开发了 hive，后来开源给了 Apache 基金会组织。 hive 是一种用 SQL 语句来协助读写、管理存储在 HDFS 上的大数据集的数据仓库软件。

hive 特点

▪ hive 最大的特点是通过类 SQL 来分析大数据，而避免了写 mapreduce Java 程序来分析数据，这样使得分析数据更容易。▪数据是存储在 HDFS 上的，hive 本身并不提供数据的存储功能▪hive 是将数据映射成数据库和一张张的表，库和表的元数据信息一般存在关系型数据库上（比如 MySQL）。▪数据存储方面：他能够存储很大的数据集，并且对数据完整性、格式要求并不严格。▪数据处理方面：不适用于实时计算和响应，使用于离线分析。

hive 基本语法

hive 中的语句有点类似于 mysql 语句，主要分为 DDL，DML，DQL，在此笔者就不做详细介绍了，略微指出一点 hive 表分为内部表、外部表、分区表和分桶表，大数据培训感兴趣的读者可以查阅相关资料。

Hive 原理

hive 架构图

hive 内核

hive 的内核是驱动引擎，驱动引擎由四部分组成，这四部分分别是：▪解释器：解释器的作用是将 hiveSQL 语句转换为语法树（AST）。▪编译器：编译器是将语法树编译为逻辑执行计划。▪优化器：优化器是对逻辑执行计划进行优化。▪执行器：执行器是调用底层的运行框架执行逻辑执行计划。

hive 底层存储

hive 的数据是存储在 HDFS 上，hive 中的库和表可以看做是对 HDFS 上数据做的一个映射，所以 hive 必须运行在一个 hadoop 集群上。

hive 程序执行过程

hive 执行器是将最终要执行的 mapreduce 程序放到 YARN 上以一系列 job 的方式去执行。

hive 元数据存储

hive 的元数据是一般是存储在 MySQL 这种关系型数据库上的，hive 与 MySQL 之间通过 MetaStore 服务交互。

表 2 hive 元数据相关信息

 hive 客户端

hive 有很多种客户端，下边简单列举了几个：▪ cli 命令行客户端：采用交互窗口，用 hive 命令行和 hive 进行通信。▪ hiveServer2 客户端：用 Thrift 协议进行通信，Thrift 是不同语言之间的转换器，是连接不同语言程序间的协议，通过 JDBC 或者 ODBC 去访问 hive(这个是目前 hive 官网推荐使用的连接方式)。▪ HWI 客户端：hive 自带的客户端，但是比较粗糙，一般不用。▪ HUE 客户端：通过 Web 页面来和 hive 进行交互，使用比较，一般在 CDH 中进行集成。

Hive 调优

hadoop 就像吞吐量巨大的轮船，启动开销大，如果每次只做小数量的输入输出，利用率将会很低。所以用好 hadoop 的首要任务是增大每次任务所搭载的数据量。hive 优化时，把 hive Sql 当做 mapreduce 程序来读，而不是当做 SQL 来读。hive 优化这要从三个层面进行，分别是基于 mapreduce 优化、hive 架构层优化和 hiveQL 层优化。

基于 mapreduce 优化

合理设置 map 数

上面的 mapreduce 执行过程部分介绍了，在执行 map 函数之前会先将 HDFS 上文件进行分片，得到的分片做为 map 函数的输入，所以 map 数量取决于 map 的输入分片(inputsplit)，一个输入分片对应于一个 map task，输入分片由三个参数决定，如表 3：

表 3 输入分片决定参数

公式：分片大小=max(mapreduce.min.split.size,min(dfs.block.size, mapreduce.max.split.size)),默认情况下分片大小和 dfs.block.size 是一致的，即一个 HDFS 数据块对应一个输入分片，对应一个 map task。这时候一个 map task 中只处理一台机器上的一个数据块，不需要将数据跨网络传输，提高了数据处理速度。

合理设置 reduce 数

决定 reduce 数量的相关参数如表 4：

表 4 reduce 数决定参数

所以可以用 set mapred.reduce.tasks 手动调整 reduce task 个数。

hive 架构层优化

不执行 mapreduce

hive 从 HDFS 读取数据，有两种方式：启用 mapreduce 读取、直接抓取。

set hive.fetch.task.conversion=more

hive.fetch.task.conversion 参数设置成 more，可以在 select、where 、limit 时启用直接抓取方式，能明显提升查询速度。

本地执行 mapreduce

hive 在集群上查询时，默认是在集群上 N 台机器上运行，需要多个机器进行协调运行，这个方式很好地解决了大数据量的查询问题。但是当 hive 查询处理的数据量比较小时，其实没有必要启动分布式模式去执行，因为以分布式方式执行就涉及到跨网络传输、多节点协调等，并且消耗资源。这个时间可以只使用本地模式来执行 mapreduce job，只在一台机器上执行，速度会很快。

JVM 重用

因为 hive 语句最终要转换为一系列的 mapreduce job 的，而每一个 mapreduce job 是由一系列的 map task 和 Reduce task 组成的，默认情况下，mapreduce 中一个 map task 或者一个 Reduce task 就会启动一个 JVM 进程，一个 task 执行完毕后，JVM 进程就退出。这样如果任务花费时间很短，又要多次启动 JVM 的情况下，JVM 的启动时间会变成一个比较大的消耗，这个时候，就可以通过重用 JVM 来解决。

set mapred.job.reuse.jvm.num.tasks=5

这个设置就是制定一个 jvm 进程在运行多次任务之后再退出，这样一来，节约了很多的 JVM 的启动时间。

并行化

一个 hive sql 语句可能会转为多个 mapreduce job，每一个 job 就是一个 stage，这些 job 顺序执行，这个在 hue 的运行日志中也可以看到。但是有时候这些任务之间并不是是相互依赖的，如果集群资源允许的话，可以让多个并不相互依赖 stage 并发执行，这样就节约了时间，提高了执行速度，但是如果集群资源匮乏时，启用并行化反倒是会导致各个 job 相互抢占资源而导致整体执行性能的下降。启用并行化：

set hive.exec.parallel=true

hiveQL 层优化

利用分区表优化

分区表是在某一个或者某几个维度上对数据进行分类存储，一个分区对应于一个目录。在这中的存储方式，当查询时，如果筛选条件里有分区字段，那么 hive 只需要遍历对应分区目录下的文件即可，不用全局遍历数据，使得处理的数据量大大减少，提高查询效率。当一个 hive 表的查询大多数情况下，会根据某一个字段进行筛选时，那么非常适合创建为分区表。

利用桶表优化

桶表的概念在本教程第一步有详细介绍，就是指定桶的个数后，存储数据时，根据某一个字段进行哈希后，确定存储再哪个桶里，这样做的目的和分区表类似，也是使得筛选时不用全局遍历所有的数据，只需要遍历所在桶就可以了。

hive.optimize.bucketmapJOIN=true;hive.input.format=org.apache.hadoop.hive.ql.io.bucketizedhiveInputFormat; hive.optimize.bucketmapjoin=true; hive.optimize.bucketmapjoin.sortedmerge=true; 

join 优化

▪ 优先过滤后再 join，最大限度地减少参与 join 的数据量。▪ 小表 join 大表原则应该遵守小表 join 大表原则，原因是 join 操作的 reduce 阶段，位于 join 左边的表内容会被加载进内存，将条目少的表放在左边，可以有效减少发生内存溢出的几率。join 中执行顺序是从做到右生成 job，应该保证连续查询中的表的大小从左到右是依次增加的。▪ join on 条件相同的放入一个 jobhive 中，当多个表进行 join 时，如果 join on 的条件相同，那么他们会合并为一个 mapreduce job，所以利用这个特性，可以将相同的 join on 的放入一个 job 来节省执行时间。

select pt.page_id,count(t.url) PV from rpt_page_type pt join (  select url_page_id,url from trackinfo where ds='2016-10-11' ) t on pt.page_id=t.url_page_id join (  select page_id from rpt_page_kpi_new where ds='2016-10-11' ) r on t.url_page_id=r.page_id group by pt.page_id;

▪ 启用 mapjoinmapjoin 是将 join 双方比较小的表直接分发到各个 map 进程的内存中，在 map 进程中进行 join 操作，这样就省掉了 reduce 步骤，提高了速度。▪ 桶表 mapjoin 当两个分桶表 join 时，如果 join on 的是分桶字段，小表的分桶数时大表的倍数时，可以启用 map join 来提高效率。启用桶表 mapjoin 要启用 hive.optimize.bucketmapjoin 参数。▪ Group By 数据倾斜优化 Group By 很容易导致数据倾斜问题，因为实际业务中，通常是数据集中在某些点上，这也符合常见的 2/8 原则，这样会造成对数据分组后，某一些分组上数据量非常大，而其他的分组上数据量很小，而在 mapreduce 程序中，同一个分组的数据会分配到同一个 reduce 操作上去，导致某一些 reduce 压力很大，其他的 reduce 压力很小，这就是数据倾斜，整个 job 执行时间取决于那个执行最慢的那个 reduce。解决这个问题的方法是配置一个参数：set hive.groupby.skewindata=true。当选项设定为 true，生成的查询计划会有两个 MR job。第一个 MR job 中，map 的输出结果会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 GroupBy Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。▪ Order By 优化因为 order by 只能是在一个 reduce 进程中进行的，所以如果对一个大数据集进行 order by，会导致一个 reduce 进程中处理的数据相当大，造成查询执行超级缓慢。▪ 一次读取多次插入▪ Join 字段显示类型转换▪ 使用 orc、parquet 等列式存储格式

原创作者：bling ss

​