使用Hive SQL插入动态分区的Parquet表OOM异常分析

一、异常描述

当运行“INSERT … SELECT”语句向Parquet或者ORC格式的表中插入数据时，如果启用了动态分区，你可能会碰到以下错误，而导致作业无法正常执行。

Hive客户端：

Task with the most failures(4):
Diagnostic Messages for this Task:
Error: GC overhead limit exceeded
...
FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask
MapReduce Jobs Launched:
Stage-Stage-1: Map: 1 HDFS Read: 0 HDFS Write: 0 FAIL
Total MapReduce CPU Time Spent: 0 msec

YARN的8088中查看具体map task报错：

1	2017-10-27 17:08:04,317 FATAL [main] org.apache.hadoop.mapred.YarnChild: Error running child : java.lang.OutOfMemoryError: GC overhead limit exceeded

二、异常分析
Parquet和ORC是列式批处理文件格式。这些格式要求在写入文件之前将批次的行（batches of rows）缓存在内存中。在执行INSERT语句时，动态分区目前的实现是：至少为每个动态分区目录打开一个文件写入器（file writer）。由于这些缓冲区是按分区维护的，因此在运行时所需的内存量随着分区数量的增加而增加。所以经常会导致mappers或reducers的OOM，具体取决于打开的文件写入器（file writer）的数量。

通过INSERT语句插入数据到动态分区表中，也可能会超过HDFS同时打开文件数的限制。

如果没有join或聚合，INSERT … SELECT语句会被转换为只有map任务的作业。mapper任务会读取输入记录然后将它们发送到目标分区目录。在这种情况下，每个mapper必须为遇到的每个动态分区创建一个新的文件写入器（file writer）。mapper在运行时所需的内存量随着它遇到的分区数量的增加而增加。

三、异常重现与解决
3.1. 生成动态分区的几个参数说明

hive.exec.dynamic.partition

默认值：false
是否开启动态分区功能，默认false关闭。
使用动态分区时候，该参数必须设置成true;

hive.exec.dynamic.partition.mode

默认值：strict
动态分区的模式，默认strict，表示必须指定至少一个分区为静态分区，nonstrict模式表示允许所有的分区字段都可以使用动态分区。
一般需要设置为nonstrict

hive.exec.max.dynamic.partitions.pernode

默认值：100
在每个执行MR的节点上，最大可以创建多少个动态分区。
该参数需要根据实际的数据来设定。
比如：源数据中包含了一年的数据，即day字段有365个值，那么该参数就需要设置成大于365，如果使用默认值100，则会报错。

hive.exec.max.dynamic.partitions
默认值：1000
在所有执行MR的节点上，最大一共可以创建多少个动态分区。
同上参数解释。

hive.exec.max.created.files

默认值：100000
整个MR Job中，最大可以创建多少个HDFS文件。
一般默认值足够了，除非你的数据量非常大，需要创建的文件数大于100000，可根据实际情况加以调整。

mapreduce.map.memory.mb
map任务的物理内存分配值，常见设置为1GB,2GB,4GB等。

mapreduce.map.java.opts

map任务的Java堆栈大小设置，一般设置为小于等于上面那个值的75%，这样可以保证map任务有足够的堆栈外内存空间。

mapreduce.input.fileinputformat.split.maxsize
mapreduce.input.fileinputformat.split.minsize

这个两个参数联合起来用，主要是为了方便控制mapreduce的map数量。比如我设置为1073741824，就是为了让每个map处理1GB的文件。

3.2. 一个例子
前两天给人调一个使用Hive SQL插入动态分区的Parquet表时，总是报错OOM，也是折腾了很久。以下我们来看看整个过程。

1. 首先我们看看执行脚本的内容，基本其实就是使用Hive的insert语句将文本数据表插入到另外一张parquet表中，当然使用了动态分区。

2. 我们看看原始数据文件，是文本文件，一共120个，每个30GB大小，总共差不多3.6TB。

3. 我们看看报错

4. 因为是一个只有map的mapreduce任务，当我们从YARN的8088观察这个作业时可以发现，基本没有一个map能够执行成功，全部都是失败的。报上面的错误。

5. 把mapreduce.map.memory.mb从2GB增大到4GB，8GB，16GB，相应mapreduce.map.java.opts增大到3GB，6GB，12GB。依旧报错OOM。

6. 后面又将mapreduce.input.fileinputformat.split.maxsize从1GB，减少为512MB，256MB，从而增大map数量，缩小单个map处理文件的大小。依旧报错OOM。

7. 最后启用 hive.optimize.sort.dynamic.partition，增加reduce过程，作业执行成功。

8. 最后查看结果文件大约1.2TB，约为输入文件的三分之一。一共1557个分区，最大的分区文件为2GB。

四、异常总结
对于这个异常，我们建议有以下三种方式来处理：

启用 hive.optimize.sort.dynamic.partition=true;

通过这个优化，这个只有map任务的mapreduce会引入reduce过程，这样动态分区的那个字段比如日期在传到reducer时会被排序。由于分区字段是排序的，因此每个reducer只需要保持一个文件写入器（file writer）随时处于打开状态，在收到来自特定分区的所有行后，关闭记录写入器（record writer），从而减小内存压力。这种优化方式在写parquet文件时使用的内存要相对少一些，但代价是要对分区字段进行排序。
第二种方式就是增加每个mapper的内存分配

即增大mapreduce.map.memory.mb和mapreduce.map.java.opts，这样所有文件写入器（filewriter）缓冲区对应的内存会更充沛。
将查询分解为几个较小的查询，以减少每个查询创建的分区数量。这样可以让每个mapper打开较少的文件写入器（file writer）。

备注：
默认情况下，Hive为每个打开的Parquet文件缓冲区（file buffer）分配128MB。这个buffer大小由参数parquet.block.size控制。为获得最佳性能，parquet的buffer size需要与HDFS的block size保持对齐（比如相等），从而使每个parquet文件在单个HDFS的块中，以便每个I/O请求都可以读取整个数据文件，而无需通过网络传输访问后续的block。