关于hive:hive问题处理

内存溢出

Client 端内存溢出

Client 端产生内存溢出执行上面的看是很简略的一条 sql 语句：
hive> select count(1) from test_tb_1_1;

Query ID = hdfs_20180802104347_615d0836-cf41-475d-9bec-c62a1f408b21
Total jobs = 1
Launching Job 1 out of 1
Number of reduce tasks determined at compile time: 1
In order to change the average load for a reducer (in bytes):
set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
set mapreduce.job.reduces=<number>
FAILED: Execution Error, return code -101 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask. Java heap space

报错起因：
该语句会进行全表全分区扫描，如果该表的分区数很多，数据量很大，可能就会呈现客户端内存不足的报错。
注：客户端报内存溢出的判断根据，通过查看客户端输入来的日志中，还没有打印出作业的 application id 信息（信息款式如下)就报内存溢出的异样了，在 ResourceManager 上也查看不到该作业的任何信息。

因为是客户端，在启动 hive 的时候就要指定好参数，启动之后批改不了，因而须要在启动 hive 命令之前，先批改环境变量
export HIVE_CLIENT_OPTS=”-Xmx1536m -XX:MaxDirectMemorySize=512m”
默认值是 1.5G , 能够依据须要调大一些。例如：
export HIVE_CLIENT_OPTS=”-Xmx2g -XX:MaxDirectMemorySize=512m”

ApplicationMaster 端内存溢出

为了演示，先将 am 的内存调小到 512M。之后客户端输出如下的报错信息：

set yarn.app.mapreduce.am.resource.mb=512;
select count(1) from (select num, count(1) from default.test_tb_1_1 where part>180 and part<190 group by num) a;

Query ID = hdfs_20180802155500_744b90db-8a64-460f-a5cc-1428ae61931b
Total jobs = 2
Launching Job 1 out of 2
Number of reduce tasks not specified. Estimated from input data size: 849
In order to change the average load for a reducer (in bytes):
set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
set mapreduce.job.reduces=<number>
Starting Job = job_1533068988045_266336, Tracking URL = http://hadoop-jrsy-rm02.pekdc1.jdfin.local:8088/proxy/application_1533068988045_266336/
Kill Command = /soft/home/hadoop/bin/hadoop job -kill job_1533068988045_266336
Hadoop job information for Stage-1: number of mappers: 0; number of reducers: 0
2018-08-02 15:55:48,910 Stage-1 map = 0%, reduce = 0%
Ended Job = job_1533068988045_266336 with errors
Error during job, obtaining debugging information...
FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask
MapReduce Jobs Launched:
Stage-Stage-1: HDFS Read: 0 HDFS Write: 0 FAIL
Total MapReduce CPU Time Spent: 0 msec

从报错信息中能够看出有 application id 信息，阐明工作提交到了 ResourceManager 上，不是 client 端导致的，然而有没有看到 container 容器失败的信息，间接 0% 后就失败，初步能够判断是因为 application master 内存不足导致的，
进一步确定须要两种路径都行：
1）在 ResourceManager 的界面查看程序的运行信息，能够看到如下的报错信息：

Current usage: 348.2 MB of 512 MB physical memory used; 5.2 GB of 1.5 GB virtual memory used. Killing container.
阐明内存溢出了，Application application_1533068988045_266336 failed 2 times due to AM Container for appattempt_1533068988045_266336_000002 exited with exitCode: -103
能够确定是 application master 产生的内存溢出。
2）能够应用 yarn 的命令来获取日志信息
yarn logs -applicationId application_1533068988045_266336 > application_1533068988045_266336.log
该日志信息中的数据和下面的一样，判断形式雷同。注：应用该命令要留神日志文件很大的作业，因而最好不在系统盘做这件事。
解决办法：
application master 内存溢出设置上面的参数（单位 MB）

set yarn.app.mapreduce.am.resource.mb=4096;
set yarn.app.mapreduce.am.command-opts=-Xmx3276M; 注：java.opts 是 memory.mb 的 80% 左右
set yarn.app.mapreduce.am.resource.cpu-vcores=8; –MR ApplicationMaster 占用的虚构 CPU 个数

Map 端 和 Reduce 端内存溢出

Map 端 和 Reduce 端呈现内存溢出的标记：
首先会看到提醒 container 内存不足的信息 (例如：beyond physical memory)：
Container [pid=192742,containerID=container_e1383_1533068988045_250824_01_003156] is running beyond physical memory limits. Current usage: 2.0 GB of 2 GB physical memory used; 3.7 GB of 6.2 GB virtual memory used. Killing container.
通过日志信息中 attempt 信息 attempt_1533068988045_250824_r_000894_3 两头的 r 和 m 能够看出是 map 还是 reduce 容器挂了。
此处的谬误日志是 reduce 容器挂了。

Task with the most failures(4):
Task ID:task_1533068988045_250824_r_000894
URL:
http://0.0.0.0:8088/taskdetails.jsp?jobid=job_1533068988045_250824&tipid=task_1533068988045_250824_r_000894
Diagnostic Messages for this Task:
Container [pid=192742,containerID=container_e1383_1533068988045_250824_01_003156] is running beyond physical memory limits. Current usage: 2.0 GB of 2 GB physical memory used; 3.7 GB of 6.2 GB virtual memory used. Killing container.

解决形式：在执行 sql 语句前 批改如下的参数(单位 MB)：
Map 容器内存溢出设置上面的参数

set mapreduce.map.memory.mb=4096;
set mapreduce.map.java.opts=-Xmx3276M;
Reduce 容器内存溢出设置上面的参数

set mapreduce.reduce.memory.mb=4096;
set mapreduce.reduce.java.opts=-Xmx3276M; 注：java.opts 是 memory.mb 的 80% 左右依据内存溢出时提醒的内存大小（例如：Current usage: 2.0 GB of 2 GB physical memory used）进行增大内存，增大的最小单位是 1GB。

参数介绍:
mapreduce.map.memory.mb: 一个 Map Task 可应用的内存下限（单位:MB），默认为 1024。如果 Map Task 理论应用的资源量超过该值，则会被强制杀死；
mapreduce.map.java.opts: Map Task 的 JVM 参数，你能够在此配置默认的 java heap size 参数，个别设置成上一个参数的 70%-80%.
mapreduce.reduce.memory.mb: 一个 Reduce Task 可应用的资源下限（单位:MB），默认为 1024。如果 Reduce Task 理论应用的资源量超过该值，则会被强制杀死；
mapreduce.reduce.java.opts: Reduce Task 的 JVM 参数，你能够在此配置默认的 java heap size 参数，个别设置成上一个参数的 70%-80%.
mapreduce.map.cpu.vcores
mapreduce.reduce.cpu.vcores

数据歪斜

景象

简略来说数据歪斜就是数据的 key 的分化重大不均，造成一部分数据很多，一部分数据很少的场面。
举个 word count 的入门例子，它的 map 阶段就是造成（“aaa”,1）的模式，而后在 reduce 阶段进行 value 相加，得出“aaa”呈现的次数。若进行 word count 的文本有 100G，其中 80G 全副是“aaa”剩下 20G 是其余单词，那就会造成 80G 的数据量交给一个 reduce 进行相加，其余 20G 依据 key 不同扩散到不同 reduce 进行相加的状况。如此就造成了数据歪斜，临床反馈就是 reduce 跑到 99% 而后始终在原地等着 那 80G 的 reduce 跑完。

常见容易呈现数据歪斜的操作

join

参数：

1. mapjoin

set hive.auto.convert.join=true; 依据输出文件的大小决定是否将一般 join 转换为 mapjoin 的一种优化，默认不开启 false；
set hive.mapjoin.smalltable.filesize=50000000; # The threshold for the input file size of the small tables; if the file size is smaller than # this threshold, it will try to convert the common join into map join

1. Skew Join

set hive.optimize.skewjoin=true;– 如果是 join 过程呈现歪斜 应该设置为 true
set hive.skewjoin.key=100000; – 这个是 join 的键对应的记录条数超过这个值则会进行分拆, 值依据具体数据量设置。hive 在运行的时候没有方法判断哪个 key 会产生多大的歪斜，所以应用这个参数管制歪斜的阈值，如果超过这个值，新的值会发送给那些还没有达到的 reduce, 个别能够设置成（解决的总记录数 /reduce 个数）的 2 - 4 倍都能够承受.

场景：
1》大表和小表关联时
比方，一个上千万行的记录表和一个几千行表之间 join 关联时，容易产生数据歪斜。为什么大表和小表容易产生数据歪斜 (无非有的 reduce 执行工夫被提早)？可参考 Hive 中小表与大表关联(join) 的性能剖析

解决形式：
1) 多表关联时，将小表 (关联键记录少的表) 顺次放到后面，这样能够触发 reduce 端更少的操作次数，缩小运行工夫。
2) 同时能够应用 Map Join 让小的维度表缓存到内存。在 map 端实现 join 过程，从而省略掉 reduce 端的工作。然而应用这个性能，须要开启 map-side join 的设置属性：set hive.auto.convert.join=true(默认是 false)。同时还能够设置应用这个优化的小表的大小：set hive.mapjoin.smalltable.filesize=25000000(默认值 25M)
3) Mapjoin，在 map 阶段实现 join，不须要 reduce. 

 Select /*+ MAPJOIN(b) */
            a.key, a.value from a join b on a.key = b.key

2》大表和大表的关联
1) 比方：大表和大表关联，然而其中一张表的多是空值或者 0 比拟多，容易 shuffle 给一个 reduce，造成运行慢。

解决办法：把空置变成一个字符串加上随机数，把歪斜的数据分布到不同的 reduce 上

select * from log a
left outer join users b
on case when a.user_id is null then concat(‘hive’,rand() ) else a.user_id end = b.user_id;

3》Join 关联，不同数据类型关联产生数据歪斜，易呈现数据歪斜
场景：用户表中 user_id 字段为 int，log 表中 user_id 字段既有 string 类型也有 int 类型。当依照 user_id 进行两个表的 Join 操作时，默认的 Hash 操作会按 int 型的 id 来进行调配，这样会导致所有 string 类型 id 的记录都调配到一个 Reducer 中。

能够思考把数据类型转化为同一数据类型

select * from user a 
left outer join log b
on a.user_id=cast(b.user_id as string);

4》join 防止多对多关联
在 join 链接查问时，确认是否存在多对多的关联，起码保障有一个表的后果集的关联字段不反复

5》join 防止笛卡尔乘积
在 join 链接查问中没有 on 链接，而通过 where 条件语句会产生笛卡尔集

问题：Select a.key, b.value from A a join B b where a.key=b.key
优化：Select a.key, b.value from A a join B b on a.key=b.key

groupby

group by 维度过小，某值的数量过多

set hive.map.aggr=true；
set hive.groupby.skewindata=true;

hive.map.aggr=true：
在 map 中会做局部汇集操作，效率更高但须要更多的内存。
hive.groupby.skewindata=true：
数据歪斜时负载平衡，当选项设定为 true，生成的查问打算会有两个 MRJob。
第一个 MRJob 中，Map 的输入后果汇合会随机散布到 Reduce 中，每个 Reduce 做局部聚合操作，并输入后果，这样解决的后果是雷同的 GroupBy Key 有可能被散发到不同的 Reduce 中，从而达到负载平衡的目标；第二个 MRJob 再依据预处理的数据后果依照 GroupBy Key 散布到 Reduce 中（这个过程能够保障雷同的 GroupBy Key 被散布到同一个 Reduce 中），最初实现最终的聚合操作。

count distinct

如果数据量十分大，执行如 select a,count(distinct b) from t group by a; 类型的 SQL 时，会呈现数据歪斜的问题。
解决形式：应用 sum…group by 代替。如 select a,sum(1) from (select a,b from t group by a,b) group by a;
尽量避免应用 distinct 进行排重，特地是大表操作，用 group by 代替
优化前：Select count (distinct key) from a
优化后：Select sum(1) from (Select key from a group by key) t

调整 map/reduce 个数

管制 hive 工作中的 map 数

1. 通常状况下，作业会通过 input 的目录产生一个或者多个 map 工作。

次要的决定因素有：input 的文件总个数，input 的文件大小，集群设置的文件块大小(目前为 128M, 可在 hive 中通过 set dfs.block.size; 命令查看到，该参数不能自定义批改)；

1. 举例：

a) 假如 input 目录下有 1 个文件 a, 大小为 780M, 那么 hadoop 会将该文件 a 分隔成 7 个块（6 个 128m 的块和 1 个 12m 的块），从而产生 7 个 map 数
b) 假如 input 目录下有 3 个文件 a,b,c, 大小别离为 10m，20m，130m，那么 hadoop 会分隔成 4 个块（10m,20m,128m,2m）, 从而产生 4 个 map 数
即，如果文件大于块大小(128m), 那么会拆分，如果小于块大小，则把该文件当成一个块。

1. 是不是 map 数越多越好？

答案是否定的。如果一个工作有很多小文件（远远小于块大小 128m）, 则每个小文件也会被当做一个块，用一个 map 工作来实现，而一个 map 工作启动和初始化的工夫远远大于逻辑解决的工夫，就会造成很大的资源节约。而且，同时可执行的 map 数是受限的。

1. 是不是保障每个 map 解决靠近 128m 的文件块，就居安思危了？

答案也是不肯定。比方有一个 127m 的文件，失常会用一个 map 去实现，但这个文件只有一个或者两个小字段，却有几千万的记录，如果 map 解决的逻辑比较复杂，用一个 map 工作去做，必定也比拟耗时。
针对下面的问题 3 和 4，咱们须要采取两种形式来解决：即缩小 map 数和减少 map 数；

如何合并小文件，缩小 map 数？

假如一个 SQL 工作：

 Select count(1) from popt_tbaccountcopy_mes where pt =‘2012-07-04’;
 该工作的 inputdir  /group/p_sdo_data/p_sdo_data_etl/pt/popt_tbaccountcopy_mes/pt=2012-07-04
 共有 194 个文件，其中很多是远远小于 128m 的小文件，总大小 9G，失常执行会用 194 个 map 工作。Map 总共耗费的计算资源：SLOTS_MILLIS_MAPS= 623,020
 我通过以下办法来在 map 执行前合并小文件，缩小 map 数：

    set hive.hadoop.supports.splittable.combineinputformat = true;
    set mapred.max.split.size=100000000;--- 每个 Mapper 工作解决的数据量的最大值. (这个值决定了合并后文件的大小)
    set mapred.min.split.size.per.node=100000000;--- 一个节点上 split 片的最小值(这个值决定了多个 DataNode 上的文件是否须要合并)
    set mapred.min.split.size.per.rack=100000000;--- 一个交换机下 split 片的最小值(这个值决定了多个交换机上的文件是否须要合并)
    set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

再执行下面的语句，用了 74 个 map 工作，map 耗费的计算资源：SLOTS_MILLIS_MAPS= 333,500
对于这个简略 SQL 工作，执行工夫上可能差不多，但节俭了一半的计算资源。
大略解释一下，100000000 示意 100M, set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; 这个参数示意执行前进行小文件合并，
后面三个参数确定合并文件块的大小，大于文件块大小 128m 的，依照 128m 来分隔，小于 128m, 大于 100m 的，依照 100m 来分隔，把那些小于 100m 的（包含小文件和分隔大文件剩下的）,
进行合并, 最终生成了 74 个块。

如何适当的减少 map 数？

当 input 的文件都很大，工作逻辑简单，map 执行十分慢的时候，能够思考减少 Map 数，来使得每个 map 解决的数据量缩小，从而进步工作的执行效率。
假如有这样一个工作：

 Select data_desc,
        count(1),
        count(distinct id),
        sum(case when …),
        sum(case when ...),
        sum(…)
from a group by data_desc

如果表 a 只有一个文件，大小为 120M，但蕴含几千万的记录，如果用 1 个 map 去实现这个工作，必定是比拟耗时的，这种状况下，咱们要思考将这一个文件正当的拆分成多个，
这样就能够用多个 map 工作去实现。

set mapred.reduce.tasks=10;
create table a_1 as 
select * from a 
distribute by rand(123); 

这样会将 a 表的记录，随机的扩散到蕴含 10 个文件的 a_1 表中，再用 a_1 代替下面 sql 中的 a 表，则会用 10 个 map 工作去实现。
每个 map 工作解决大于 12M（几百万记录）的数据，效率必定会好很多。
看上去，貌似这两种有些矛盾，一个是要合并小文件，一个是要把大文件拆成小文件，这点正是重点须要关注的中央，
依据理论状况，管制 map 数量须要遵循两个准则：使大数据量利用适合的 map 数；使单个 map 工作解决适合的数据量；

管制 hive 工作的 reduce 数

1. Hive 本人如何确定 reduce 数：

reduce 个数的设定极大影响工作执行效率，不指定 reduce 个数的状况下，Hive 会猜想确定一个 reduce 个数，基于以下两个设定：
hive.exec.reducers.bytes.per.reducer（每个 reduce 工作解决的数据量，默认为 1000^3=1G）
hive.exec.reducers.max（每个工作最大的 reduce 数，默认为 1009）
计算 reducer 数的公式很简略 N =min(参数 2，总输出数据量 / 参数 1) 即 min(hive.exec.reducers.max，总输出数据量 /hive.exec.reducers.bytes.per.reducer)
即，如果 reduce 的输出（map 的输入）总大小不超过 1G, 那么只会有一个 reduce 工作；
如：select pt,count(1) from popt_tbaccountcopy_mes where pt = ‘2012-07-04’ group by pt;
/group/p_sdo_data/p_sdo_data_etl/pt/popt_tbaccountcopy_mes/pt=2012-07-04 总大小为 9G 多，因而这句有 10 个 reduce

reduce 个数并不是越多越好；
同 map 一样，启动和初始化 reduce 也会耗费工夫和资源；
另外，有多少个 reduce, 就会有多少个输入文件，如果生成了很多个小文件，那么如果这些小文件作为下一个工作的输出，则也会呈现小文件过多的问题；

2. 调整 reduce 个数：

办法一：调整 hive.exec.reducers.bytes.per.reducer 参数的值；
set hive.exec.reducers.bytes.per.reducer=500000000;（500M）
select pt,count(1) from popt_tbaccountcopy_mes where pt = ‘2012-07-04’ group by pt; 这次有 20 个 reduce

办法二：调整 reduce 个数 mapred.reduce.tasks；
set mapred.reduce.tasks = 15;
select pt,count(1) from popt_tbaccountcopy_mes where pt = ‘2012-07-04’ group by pt; 这次有 15 个 reduce

3. 什么状况下只有一个 reduce；

很多时候你会发现工作中不论数据量多大，不论你有没有设置调整 reduce 个数的参数，工作中始终都只有一个 reduce 工作；
其实只有一个 reduce 工作的状况，除了数据量小于 hive.exec.reducers.bytes.per.reducer 参数值的状况外，还有以下起因：
a) 没有 group by 的汇总
比方把 select pt,count(1) from popt_tbaccountcopy_mes where pt = ‘2012-07-04’ group by pt;
写成 select count(1) from popt_tbaccountcopy_mes where pt = ‘2012-07-04’;
这点十分常见，心愿大家尽量改写。
b) 用了 Order by
c) 有笛卡尔积
通常这些状况下，除了找方法来变通和防止，我临时没有什么好的方法，因为这些操作都是全局的，所以 hadoop 不得不用一个 reduce 去实现；
同样的，在设置 reduce 个数的时候也须要思考这两个准则：使大数据量利用适合的 reduce 数；使单个 reduce 工作解决适合的数据量；

主动合并输入的小文件

set hive.merge.mapfiles = true;
set hive.merge.mapredfiles = true;
set hive.merge.size.per.task = 256000000;
set hive.merge.smallfiles.avgsize=256000000;
set hive.merge.orcfile.stripe.level=false;

hive tez insert union all 问题

（1）问题形容
在 hive 中应用 tez 模式时，发现 tez 的输入后果在对应表目录中，生成了子目录，造成未配置 tez 的 hive 客户端对该表进行读取时，无奈获取到数据。
检查表的输入目录，在分区目录下发现了两个子目录:1 和 2：
/user/hive/test1/20170920000000/1
/user/hive/test1/20170920000000/2
（2）起因

查看对应的 sql，发现存在 insert union 操作，查看往上信息，发现 tez 对于 insert union 操作会进行优化，通过并行加快速度，为避免有雷同文件输入，所以对并行的输入各自生成成了一个子目录，在子目录中寄存后果。如果此时全副 hive 客户端引擎及相干都设定为 tez，则无问题。如果有客户端还在应用 mr 引擎，则会呈现读取不到数据的状况。而在 hive 中，对于表目录寄存信息有如下策略：

在 Hive 默认设置中，hive 只能读取对应表目录下的文件，
1）如果表目录中同时存在目录和文件
则应用 hive 进行读取时，会报目录非文件的谬误
2）表目录中只有目录
hive 不会进行深层递归查问，只会读取到对应查问目录，会查问后果为空
3）表目录中只有文件
则能够失常查问

Hive on tez 中对 insert union 操作进行了优化，会并行输入到对应的表目录，为避免有雷同名文件存在，所以为各自输入在表目录下各自设置了一个目录，里边寄存执行后果
此种目录，对 tez 引擎客户端可读。然而对于 mr 引擎，因为其只会便遍历到对应分区层。发现分区下没有文件后，就返回空
（3）解决思路
发现该景象后，从网上搜寻解决方案，http://grokbase.com/t/hive/us…
经该文章提醒，能够开启 mapreduce 的递归查问模式：
set mapreduce.input.fileinputformat.input.dir.recursive=true
            在 hive 中，设置执行查问后，会提醒谬误，要求将 set hive.mapred.supports.subdirectories=true;
 通过两个 set 后，仔执行查问，hive 即可对 tez 引擎产 union 产出的数据进行拜访。

同时为了查看 tez 的此种状况对其余引擎的兼容性，又对 sparktThriftServer 的 sql 拜访进行了测试：

        对于 sparkthriftserver，只需增加 set mapreduce.input.fileinputformat.input.dir.recursive=true，即可拜访对应数据
总结一下 sparkThriftServer 模式与 hive 的不同
1）在表目录下同时有文件和目录时（非分区目录）
此时 sparkThriftServer 和 hive 都会报非文件谬误
2）在表目录下均为目录（非分区目录）
此时 sparkThriftServer 会报非文件谬误，hive 无错，然而查问为空

（4）解决方案
1）计划一：所有 hive 端都替换为 tez 引擎
这不会呈现上述问题，但会存在对其余引擎如 spqrk，mr，presto 等的影响
2）计划二：为 hadoop 和 hive 设置可递归读
该计划绝对可行，但须要调研对一般 mr 工作的影响