hive的企业级调优
1、Fetch抓取
Fetch抓取是指，Hive中对某些情况的查询可以不必使用MapReduce计算；
-
例如：select * from score;
-
在这种情况下，Hive可以简单地读取对应的存储目录下的文件，然后输出查询结果到控制台；
-
在hive-default.xml.template文件中 hive.fetch.task.conversion默认是more；
案例：
（1）把 hive.fetch.task.conversion设置成none，然后执行查询语句，都会执行mapreduce程序；
set hive.fetch.task.conversion=none;
select * from score;
select s_id from score;
select s_id from score limit 3;
（2）把hive.fetch.task.conversion设置成more，然后执行查询语句，如上三种查询方式都不会执行mapreduce程序。
2、本地模式
-
在Hive客户端测试时，默认情况下是启用hadoop的job模式,把任务提交到集群中运行，这样会导致计算非常缓慢；
-
Hive可以通过本地模式在单台机器上处理任务。对于小数据集，执行时间可以明显被缩短。
案例 ：
（1）开启本地模式，并执行查询语句
set hive.exec.mode.local.auto=true; //开启本地mr
（2）设置local mr的最大输入数据量，当输入数据量小于这个值时采用local mr的方式，默认为134217728，即128M；
set hive.exec.mode.local.auto.inputbytes.max=50000000;
（3）设置local mr的最大输入文件个数，当输入文件个数小于这个值时采用local mr的方式，默认为4；
set hive.exec.mode.local.auto.input.files.max=5;
（4）执行查询的sql语句
select * from student cluster by s_id;
3、表的优化
3.1小表、大表 join
-
将key相对分散，并且数据量小的表放在join的左边，这样可以有效减少内存溢出错误发生的几率；再进一步，可以使用map join让小的维度表（1000条以下的记录条数）先进内存。在map端完成reduce。
select count(distinct s_id) from score;
select count(s_id) from score group by s_id; 在map端进行聚合，效率更高
select count(distinct s_id) from score;
select count(1) from (
select count(1) from score group by s_id
);
3.2大表 join 大表
（1）空 key 过滤
-
join超时，是因为某些key对应的数据太多，而相同key对应的数据都会发送到相同的reducer上，从而导致内存不够。
-
分析这些异常的key，很多情况下，这些key对应的数据是异常数据，我们需要在SQL语句中进行过滤。
不过滤：
INSERT OVERWRITE TABLE jointable SELECT a.* FROM nullidtable a JOIN ori b ON a.id = b.id;
过滤：
INSERT OVERWRITE TABLE jointable SELECT a.* FROM (SELECT * FROM nullidtable WHERE id IS NOT NULL ) a JOIN ori b ON a.id = b.id;
（2）空 key 转换
-
某个 key 为空对应的数据很多，必须要包含在 join 的结果中，此时可以表 a 中 key 为空的字段赋一个随机的值，使得数据随机均匀地分不到不同的 reducer 上。
INSERT OVERWRITE TABLE jointable
SELECT a.* FROM nullidtable a
LEFT JOIN ori b ON CASE WHEN a.id IS NULL THEN 'hive' ELSE a.id END = b.id;
后果：所有为null值的id全部都变成了相同的字符串，极其容易造成数据的倾斜（所有的key相同，相同key的数据会到同一个reduce当中去）；
解决办法：通过hive的rand函数（即：concat('hive', rand())），随记的给每一个为空的id赋上一个随机值，可以避免造成数据倾斜：
INSERT OVERWRITE TABLE jointable
SELECT a.* FROM nullidtable a
LEFT JOIN ori b ON CASE WHEN a.id IS NULL THEN concat('hive', rand()) ELSE a.id END = b.id;
3.3map join
如果不指定MapJoin 或者不符合 MapJoin的条件，那么Hive解析器会将Join操作转换成Common Join，即：在Reduce阶段完成join，容易发生数据倾斜。可以用 MapJoin 把小表全部加载到内存，在map端进行join，避免reducer处理。
（1）开启MapJoin参数设置
set hive.auto.convert.join = true; --默认为true
（2）大表小表的阈值设置（默认25M一下认为是小表）
set hive.mapjoin.smalltable.filesize=26214400;
（3）MapJoin工作机制
- （1）首先是Task A，它是一个Local Task（在客户端本地执行的Task），负责扫描小表b的数据，将其转换成一个HashTable的数据结构，并写入本地的文件中，之后将该文件加载到DistributeCache中。
- （2）接下来是Task B，该任务是一个没有Reduce的MR，启动MapTasks扫描大表a,在Map阶段，根据a的每一条记录去和DistributeCache中b表对应的HashTable关联，并直接输出结果。
- （3）由于MapJoin没有Reduce，所以由Map直接输出结果文件，有多少个Map Task，就有多少个结果文件。
例：
（1）开启Mapjoin功能
set hive.auto.convert.join = true; 默认为true
（2）执行小表JOIN大表语句
INSERT OVERWRITE TABLE jointable2
SELECT b.id, b.time, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
FROM smalltable s
JOIN bigtable b
ON s.id = b.id;
（3）执行大表JOIN小表语句
INSERT OVERWRITE TABLE jointable2
SELECT b.id, b.time, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
FROM bigtable b
JOIN smalltable s
ON s.id = b.id;
3.4group By
-
默认情况下，Map阶段同一Key数据分发给一个reduce，当一个key数据过大时就倾斜了。
-
并不是所有的聚合操作都需要在Reduce端完成，很多聚合操作都可以先在Map端进行部分聚合，最后在Reduce端得出最终结果。
开启Map端聚合参数设置：
set hive.map.aggr = true;
set hive.groupby.mapaggr.checkinterval = 100000;--在Map端进行聚合操作的条目数目
set hive.groupby.skewindata = true;--有数据倾斜的时候进行负载均衡（默认是false）
当上述选项设定为 true，生成的查询计划会有两个MR Job。
- 第一个MR Job中，Map的输出结果会随机分布到Reduce中，每个Reduce做部分聚合操作，并输出结果，这样处理的结果是相同的Group By Key有可能被分发到不同的Reduce中，从而达到负载均衡的目的；
- 第二个MR Job再根据预处理的数据结果按照Group By Key分布到Reduce中（这个过程可以保证相同的Group By Key被分布到同一个Reduce中），最后完成最终的聚合操作。
3.5count(distinct)
数据量大的情况下，由于count distinct 操作需要用一个reduce Task来完成，这一个Reduce需要处理的数据量太大，就会导致整个Job很难完成，一般count distinct使用先group by 再count的方式替换。
如：select count(ip) from (select ip from log_text group by ip) t;
3.6笛卡尔积
-
尽量避免笛卡尔积，即避免join的时候不加on条件，或者无效的on条件
-
Hive只能使用1个reducer来完成笛卡尔积。
4、使用分区剪裁、列剪裁
-
尽可能早地过滤掉尽可能多的数据量，避免大量数据流入外层SQL。
-
列剪裁
-
只获取需要的列的数据，减少数据输入。
-
-
分区裁剪
-
分区在hive实质上是目录，分区裁剪可以方便直接地过滤掉大部分数据。
-
尽量使用分区过滤，少用select *；
-
- 逻辑条件写在ON后面：先Where再关联 ：
SELECT a.id FROM ori a LEFT JOIN bigtable b ON (a.id <= 10 AND a.id = b.id);
或者直接写成子查询：
SELECT a.id FROM bigtable a RIGHT JOIN (SELECT id FROM ori WHERE id <= 10) b ON a.id = b.id;
5、并行执行
-
把一个sql语句中没有相互依赖的阶段并行去运行。提高集群资源利用率
-
--开启并行执行
set hive.exec.parallel=true;
--同一个sql允许最大并行度，默认为8。
set hive.exec.parallel.thread.number=16;
6、严格模式
-
Hive提供了一个严格模式，可以防止用户执行那些可能意想不到的不好的影响的查询。
-
通过设置属性hive.mapred.mode值为默认是非严格模式nonstrict 。开启严格模式需要修改hive.mapred.mode值为strict，开启严格模式可以禁止如下3种类型的查询。
- 设置非严格模式（默认）
set hive.mapred.mode=nonstrict;
- 设置严格模式
set hive.mapred.mode=strict;
（1）严格模式下，对于分区表，除非where语句中含有分区字段过滤条件来限制范围，否则不允许执行；
（2）对于使用了order by语句的查询，要求必须使用limit语句；
（3）限制笛卡尔积的查询
-
严格模式下，避免出现笛卡尔积的查询；
7、JVM重用
-
JVM重用是Hadoop调优参数的内容，其对Hive的性能具有非常大的影响，特别是对于很难避免小文件的场景或task特别多的场景，这类场景大多数执行时间都很短。
Hadoop的默认配置通常是使用派生JVM来执行map和Reduce任务的。这时JVM的启动过程可能会造成相当大的开销，尤其是执行的job包含有成百上千task任务的情况。
-
JVM重用可以使得JVM实例在同一个job中重新使用N次。N的值可以在Hadoop的mapred-site.xml文件中进行配置。通常在10-20之间，具体多少需要根据具体业务场景测试得出。
|
<property> |
也可以在hive当中通过 set mapred.job.reuse.jvm.num.tasks=10;
8、推测执行
Hadoop采用了推测执行（Speculative Execution）机制，它根据一定的法则推测出“拖后腿”的任务，并为这样的任务启动一个备份任务，让该任务与原始任务同时处理同一份数据，并最终选用最先成功运行完成任务的计算结果作为最终结果。
设置开启推测执行参数：Hadoop的mapred-site.xml文件中进行配置：
|
<property> <property> |
hive本身也提供了配置项来控制reduce-side的推测执行：
| <property> <name>hive.mapred.reduce.tasks.speculative.execution</name> <value>true</value> <description>Whether speculative execution for reducers should be turned on. </description> </property> |
9、压缩
参见数据的压缩，Hive表中间数据压缩
#设置为true为**中间数据压缩功能，默认是false，没有开启
set hive.exec.compress.intermediate=true;
#设置中间数据的压缩算法
set mapred.map.output.compression.codec= org.apache.hadoop.io.compress.SnappyCodec;
10、使用EXPLAIN（查看hql执行计划）
11、数据倾斜的解决办法
11.1 合理设置Map数
1) 通常情况下，作业会通过input的目录产生一个或者多个map任务。
-
主要的决定因素有：input的文件总个数，input的文件大小，集群设置的文件块大小。
举例：
a) 假设input目录下有1个文件a，大小为780M，那么hadoop会将该文件a分隔成7个块（6个128m的块和1个12m的块），从而产生7个map数。
b) 假设input目录下有3个文件a，b，c大小分别为10m，20m，150m，那么hadoop会分隔成4个块（10m，20m，128m，22m），从而产生4个map数。即，如果文件大于块大小(128m)，那么会拆分，如果小于块大小，则把该文件当成一个块。
2） 是不是map数越多越好？ （减少map数）
-
不是。如果一个任务有很多小文件（远远小于块大小128m），则每个小文件也会被当做一个块，用一个map任务来完成，而一个map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费。而且，同时可执行的map数是受限的。
3） 是不是保证每个map处理接近128m的文件块，就高枕无忧了？ （增加map数）
-
不一定。比如有一个127m的文件，正常会用一个map去完成，但这个文件只有一个或者两个小字段，却有几千万的记录，如果map处理的逻辑比较复杂，用一个map任务去做，肯定也比较耗时。
总之：针对上面的问题2和3，我们需要采取两种方式来解决：即减少map数和增加map数；
11.2 小文件合并
-
在map执行前合并小文件，减少map数：
-
CombineHiveInputFormat 具有对小文件进行合并的功能（系统默认的格式）
-
set mapred.max.split.size=112345600;
-
set mapred.min.split.size.per.node=112345600;
-
set mapred.min.split.size.per.rack=112345600;
-
set hive.input.format= org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
-
这个参数表示执行前进行小文件合并，前面三个参数确定合并文件块的大小，大于文件块大小128m的，按照128m来分隔，小于128m，大于100m的，按照100m来分隔，把那些小于100m的（包括小文件和分隔大文件剩下的），进行合并。
11.3 复杂文件增加Map数
-
当input的文件都很大，任务逻辑复杂，map执行非常慢的时候，可以考虑增加Map数，来使得每个map处理的数据量减少，从而提高任务的执行效率。
-
增加map的方法为
-
根据 computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize)))公式
-
调整maxSize最大值。让maxSize最大值低于blocksize就可以增加map的个数。
-
mapreduce.input.fileinputformat.split.minsize=1 默认值为1
mapreduce.input.fileinputformat.split.maxsize=Long.MAXValue 默认值Long.MAXValue因此，默认情况下，切片大小=blocksize
maxsize（切片最大值): 参数如果调到比blocksize小，则会让切片变小，而且就等于配置的这个参数的值。
minsize(切片最小值): 参数调的比blockSize大，则可以让切片变得比blocksize还大。
4 合理设置Reduce数
4.1调整reduce个数方法一
1）每个Reduce处理的数据量默认是256MB
set hive.exec.reducers.bytes.per.reducer=256000000;
2) 每个任务最大的reduce数，默认为1009
set hive.exec.reducers.max=1009;
3) 计算reducer数的公式
N=min(参数2，总输入数据量/参数1)
4.2调整reduce个数方法二
--设置每一个job中reduce个数
set mapreduce.job.reduces=3;
4.3reduce个数并不是越多越好
-
过多的启动和初始化reduce也会消耗时间和资源；
-
同时过多的reduce会生成很多个文件，也有可能出现小文件问题。