(13)Hive调优——动态分区导致的小文件问题

前言

  动态分区指的是：分区的字段值是基于查询结果自动推断出来的，核心语法就是insert+select。 具体内容指路文章：

https://blog.csdn.net/SHWAITME/article/details/136111924?spm=1001.2014.3001.5501文章浏览阅读483次，点赞15次，收藏8次。Hive的相关概念——分区表、分桶表https://blog.csdn.net/SHWAITME/article/details/136111924?spm=1001.2014.3001.5501

0 问题现象

现象：报错errorr如下：

[Error 20004]: Fatal error occurred when node tried to create 
too many dynamic partitions. The maximum number of dynamic 
partitions is controlled by hive.exec.max.dynamic.partitions and 
hive.exec.max.dynamic.partitions.pernode. Maximum was setto: 100

原因： Hive对其创建的动态分区数量实施限制，总结而言：每个执行MR的节点能创建动态分区的个数上限为100个(默认)，所有执行MR的节点能创建动态分区的个数上限为1000个动态分区(默认)，相关参数如下：

#在每个执行MR的节点上，最大可以创建多少个动态分区,默认值为100
hive.exec.max.dynamic.partitions.pernode=100;#在所有执行MR的节点上，最大一共可以创建多少个动态分区,默认1000
hive.exec.max.dynamic.partitions=1000;#整个MR Job中，最大可以创建多少个HDFS 文件,默认100000
hive.exec.max.created.files=100000;

  实际生产环境中，上述参数可以调整。

1 问题解决

解决方案一：调整动态分区数

set hive.exec.dynamic.partition=true;
在每个执行MR的节点上，最大可以创建256个动态分区（默认值为100）
set hive.exec.max.dynamic.partitions.pernode=256;
#在所有执行MR的节点上，最大一共可以创建2048个动态分区（默认值为1000）
set hive.exec.max.dynamic.partitions=2048;

    虽然配置了上述参数，但是不能保证小文件的问题彻底解决，有时候还需要设置reduce数。 mapred.reduce.tasks的计算公式可以为：

dynamic.partitions（总） / dynamic.partitions.pernode （分节点）<= mapred.reduce.tasks

    根据上述例子，得到 2048/256 = 8，如果mapred.reduce.tasks小于8就会报错，所以可以手动设置 set mapred.reduce.tasks=10;

方案一弊端：小文件剧增

   上述方案增加了动态分区的数量，虽然暂时不报错了，但是引出更棘手的问题，动态分区会产生大量小文件，因为当整个MR  job启动K个reduce Instance，N个目标分区，极端情况下会产生K* N个小文件。整个MR Job中，默认创建hdfs文件数的上限为100000个（参数hive.exec.max.created.files = 100000）。

     假设输入的数据量为1T，我们开启了2000 个MapReduce任务去读取，假设动态分区数总数为100个，也就是说：hdfs上一共有100个分区，每个分区下的小文件数量都是2000个。此时小文件数量=ReduceTask数量 * 分区数，即2000*100=200000个，
直接超出创建hdfs文件数的上限数（参数hive.exec.max.created.files = 100000）。例如生产环境执行下列sql进行数据插入时，动态分区会有产生小文件的风险：

insert overwrite table testA partition(dt)
select * 
from testB

  那么动态分区造成小文件应该如何避免和优化呢？

解决方案二：distribute by

    distribute by 是用来解决数据分发问题的，根据指定的分区字段值，可以控制数据分发到对应的reduce中去【HASH的方式，类似于spark中的repartition】。分区编号 =分区字段值的hash值 % reduce数，即【distribute by dt】 操作可以将同一分区的数据直接发到同一个reduce中。

   执行sql后，由原来100个分区，每个分区下2000个小文件的局面改造成：100个分区，每个分区下只有一个文件。相关sql如下：

insert overwrite table test partition(dt)
select * 
from table
distribute by dt

方案二弊端：数据倾斜

    经过上述操作，又引来了一个新的问题，假设这100个分区的数据分布不均匀的，有的redcue数据很多有几百个G，有的只有几兆，这样导致个别reduce会卡在99%，拖慢整体的HQL执行效率。因此可以采用随机数，将数据相对均衡地发送到每个reducer来解决该问题，使每个reduce任务处理的数据大体一致。

解决方案三：distribute by命令

（1）设定每个reduce处理的数据量来控制hdfs上最终生成的文件数。

       假设给每个redcue任务分配10G数据量，则对于1T的数据总共会启动102个左右的reduceTask，相关sql如下：

#每个reduce处理数据量
set hive.exec.reducers.bytes.per.reducer=1024*10*1000*1000; ---10Ginsert overwrite table test partition(dt)
select * 
from table
distribute by rand()

（2）rand()函数来控制hdfs上最终生成多少个文件【强烈推荐】

 
insert overwrite table test partition(dt)
select * 
from table
distribute by cast(rand()*100 as int);#--cast(rand()*100 as int) 生成 0-100之间的随机整数

ps：通过 distribute by cast( rand() * N as int) 来控制落地文件数， 其中 cast( rand() * N as int) 可以生成0-N之间的随机整数。

ps：更多的Hive小文件问题及解决方案见文章：

Hive的小文件问题-CSDN博客文章浏览阅读409次，点赞7次，收藏12次。Hive的小文件问题https://blog.csdn.net/SHWAITME/article/details/136108785

2 思考

    Hive底层需要限制动态分区的数量的原因是？  动态分区会在短时间内创建大量的分区，可能会占用大量的资源，主要会有以下两方面的瓶颈：

• 内存方面

      在Insert数据插入场景下，每个动态目录分区写入器(File Writer)至少会打开一个文件，对于parquert或者orc格式的文件，在写入的时候会首先写到缓冲区中，而这些缓冲区是按照分区来维护的，在运行的时候所需的内存大小会随着分区数增加而累积增加。导致OOM的mapper或者reducer，可能是由于打开的文件写入器的数量。如常见的错误：Error: GC overhead limit exceeded，针对该问题，可以调整的参数有：

#增加每个mapper的内存分配，即增大mapreduce.map.memory.mb和mapreduce.map.java.opts，这样所有文件写入器（filewriter）缓冲区对应的内存会更充沛。（1）map任务的物理内存分配值，常见设置为1GB,2GB,4GB等。
mapreduce.map.memory.mb （2）map任务的Java堆栈大小设置，一般设置为<= map任务的物理内存的75%
mapreduce.map.java.opts

• 文件句柄

        如果分区数过多，那么每个分区都会打开对应的文件句柄写入数据，可能会导致系统文件句柄占用过多，影响系统其他应用运行。因此hive又提出了一个hive.exec.max.created.files参数来控制整个mr 任务的创建文件数量的上限值（默认是100000个）

3 小结

    上述阐述hive动态分区产生小文件的最佳解决方案：distribute by cast( rand() * N as int) = 【distribute by + rand随机数】，两者互相配合，控制数据相对均衡（解决数据倾斜）的发往到指定数量的reducer中，严格控制hdfs上落地文件数目。（HQL）

   但是对于使用SparkSQL的用户来说，SparkSQL中的repartition算子可以解决这一问题，repartition和distribute by的作用一致 (控制数据发往指定分区)

    spark小文件具体的解决方案待补充~

参考文章：

Hive/Spark小文件解决方案(企业级实战)

Hive Distribute by 应用之动态分区小文件过多问题优化_distribute by cast(rand() * 99 as int)-CSDN博客