Hadoop/Hive/Spark小文件处理

什么是小文件？

小文件指的是文件size比HDFS的block size小很多的文件。Hadoop适合处理少量的大文件，而不是大量的小文件。

hadoop小文件常规的处理方式

1、小文件导致的问题

首先，在HDFS中，任何block，文件或者目录在内存中均以对象的形式存储，每个对象约占150byte，如果有1000 0000个小文件，每个文件占用一个block，则namenode大约需要2G空间。如果存储1亿个文件，则namenode需要20G空间。这样namenode内存容量严重制约了集群的扩展。
其次，访问大量小文件速度远远小于访问几个大文件。HDFS最初是为流式访问大文件开发的，如果访问大量小文件，需要不断的从一个datanode跳到另一个datanode，严重影响性能。
最后，处理大量小文件速度远远小于处理同等大小的大文件的速度。每一个小文件要占用一个slot，而task启动将耗费大量时间甚至大部分时间都耗费在启动task和释放task上。

2、Hadoop自带的解决方案

对于小文件问题，Hadoop本身也提供了几个解决方案，分别为： Hadoop Archive ， Sequence
file 和 CombineFileInputFormat 。

• Hadoop Archive Hadoop Archive或者HAR，是一个高效地将小文件放入HDFS块中的文件存档工具，它能够将多 个小文件打包成一个HAR文件，这样在减少namenode内存使用的同时，仍然允许对文件进行 透明的访问。
• Sequence file sequence file由一系列的二进制key/value组成，如果为key小文件名，value为文件内容，则可 以将大批小文件合并成一个大文件。
• CombineFileInputFormat 它是一种新的inputformat，用于将多个文件合并成一个单独的split，另外，它会考虑数据的 存储位置。

Hadoop Archive

Hadoop Archive或者HAR，是一个高效地将小文件放入HDFS块中的文件存档工具，它能够将多个小文件打包成一个HAR文件，这样在减少namenode内存使用的同时，仍然允许对文件进行透明的访问。对某个目录/foo/bar下的所有小文件存档成/outputdir/ zoo.har：

hadoop archive -archiveName zoo.har -p /foo/bar /outputdir

当然，也可以指定HAR的大小(使用-Dhar.block.size)。
HAR是在Hadoop file system之上的一个文件系统，因此所有fs shell命令对HAR文件均可用，只不过是文件路径格式不一样，HAR的访问路径可以是以下两种格式：

har://scheme-hostname:port/archivepath/fileinarchive
har:///archivepath/fileinarchive(本节点)

可以这样查看HAR文件存档中的文件：

hadoop dfs -ls har:///user/zoo/foo.har

输出：

har:///user/zoo/foo.har/hadoop/dir1
har:///user/zoo/foo.har/hadoop/dir2

使用HAR时需要两点
第一，对小文件进行存档后，原文件并不会自动被删除，需要用户自己删除；
第二，创建HAR文件的过程实际上是在运行一个mapreduce作业，因而需要有一个hadoop集群运行此命令。

此外，HAR还有一些缺陷：
第一，一旦创建，Archives便不可改变。要增加或移除里面的文件，必须重新创建归档文件。
第二，要归档的文件名中不能有空格，否则会抛出异常，可以将空格用其他符号替换
(使用-Dhar.space.replacement.enable=true 和-Dhar.space.replacement参数)。
第三，存档文件不支持压缩。

一个归档后的文件，其存储结构如下图：

Sequence file

Sequence file由一系列的二进制key/value组成，如果为key小文件名，value为文件内容，则可以将大批小文件合并成一个大文件。
Hadoop-0.21.0中提供了SequenceFile，包括Writer，Reader和SequenceFileSorter类进行写，读和排序操作。

创建sequence file的过程可以使用mapreduce工作方式完成，对于index，需要改进查找算法
优缺点：对小文件的存取都比较自由，也不限制用户和文件的多少，但是该方法不能使用append方
法，所以适合一次性写入大量小文件的场景

CombineFileInputFormat

CombineFileInputFormat是一种新的inputformat，用于将多个文件合并成一个单独的split，另外，它会考虑数据的存储位置。

Hive小文件问题怎么解决？

首先，我们要弄明白两个问题：
1）哪里会产生小文件
源数据本身有很多小文件
动态分区会产生大量小文件
reduce个数越多, 小文件越多
按分区插入数据的时候会产生大量的小文件, 文件个数 = maptask个数 * 分区数

2）小文件太多造成的影响
从Hive的角度看，小文件会开很多map，一个map开一个JVM去执行，所以这些任务的初始化，启
动，执行会浪费大量的资源，严重影响性能。
HDFS存储太多小文件, 会导致namenode元数据特别大, 占用太多内存, 制约了集群的扩展。

小文件解决方案：

方法一：通过调整参数进行合并

1）在Map输入的时候, 把小文件合并

-- 每个Map最大输入大小，决定合并后的文件数
set mapred.max.split.size=256000000;
-- 一个节点上split的至少的大小，决定了多个datanode上的文件是否需要合并
set mapred.min.split.size.per.node=100000000;
-- 一个交换机下split的至少的大小，决定了多个交换机上的文件是否需要合并
set mapred.min.split.size.per.rack=100000000;
-- 执行Map前进行小文件合并
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

2）在Reduce输出的时候, 把小文件合并

-- 在map-onlyjob后合并文件，默认true
set hive.merge.mapfiles=true;
-- 在map-reducejob后合并文件，默认false
set hive.merge.mapredfiles=true;
-- 合并后每个文件的大小，默认256000000
set hive.merge.size.per.task=256000000;
-- 平均文件大小，是决定是否执行合并操作的阈值，默认16000000 sethive.merge.smallfiles.avgsize=100000000;

方法二：针对按分区插入数据的时候产生大量的小文件的问题，可以使用DISTRIBUTE BY rand() 将数据

随机分配给Reduce，这样可以使得每个Reduce处理的数据大体一致。

-- 设置每个reducer处理的大小为5个G
set hive.exec.reducers.bytes.per.reducer=5120000000;
-- 使用distributebyrand()将数据随机分配给reduce,避免出现有的文件特别大,有的文件特别小
insert overwrite table test partition(dt)select * from iteblog_tmp
DISTRIBUTEBYrand();

方法三：使用Sequencefile作为表存储格式，不要用textfile，在一定程度上可以减少小文件

方法四：使用hadoop的archive归档

-- 用来控制归档是否可用
set hive.archive.enabled=true;
-- 通知Hive在创建归档时是否可以设置父目录
set hive.archive.har.parentdir.settable=true;-- 控制需要归档文件的大小
set har.partfile.size=1099511627776;
-- 使用以下命令进行归档
ALTER TABLE srcpart ARCHIVE PARTITION(ds='2008-04-08',hr='12');
- 对已归档的分区恢复为原文件
ALTER TABLE srcpart UNARCHIVE PARTITION(ds='2008-04-08',hr='12');
-- 注意，归档的分区不能够INSERTOVERWRITE，必须先unarchive

Spark输出文件的个数，如何合并小文件？

当使用spark sql执行etl时候出现了，最终结果大小只有几百k，但是小文件一个分区可能就有上千的情况。
小文件过多的一些危害如下：

• hdfs有最大文件数限制
• 浪费磁盘资源（可能存在空文件）
• hive中进行统计，计算的时候，会产生很多个map，影响计算的速度。

解决方案如下：

方法一：通过spark的coalesce()方法和repartition()方法

val rdd2 = rdd1.coalesce(8, true) //（true表示是否shuffle）
val rdd3 = rdd1.repartition(8)

coalesce：coalesce()方法的作用是返回指定一个新的指定分区的Rdd，如果是生成一个窄依赖的结果，那么可以不发生shuffle，分区的数量发生激烈的变化，计算节点不足，不设置true可能会出错。
repartition：coalesce()方法shuffle为true的情况。

方法二：降低spark并行度，即调节spark.sql.shuffle.partitions

比如之前设置的为100，按理说应该生成的文件数为100；
但是由于业务比较特殊，采用的大量的union all，且union all在spark中属于窄依赖，不会进行shuffle，所以导致最终会生成（union all数量+1）*100的文件数。
如有10个union all，会生成1100个小文件。这样导致降低并行度为10之后，执行时长大大增加，且文件数依旧有110个，效果不理想。

方法三：新增一个并行度=1任务，专门合并小文件

先将原来的任务数据写到一个临时分区（如tmp）；再起一个并行度为1的任务，类似：

insert overwrite 目标表 select * from 临时分区

结果小文件数还是没有减少，经过多次测后发现原因：‘select * from 临时分区’ 这个任务在spark中属于窄依赖；
并且spark DAG中分为宽依赖和窄依赖，只有宽依赖会进行shuffle；
故并行度shuffle，spark.sql.shuffle.partitions=1也就没有起到作用；

由于数据量本身不是特别大，所以直接采用了group by（在spark中属于宽依赖）的方式，类似：

insert overwrite 目标表 select * from 临时分区 group by * 

先运行原任务，写到tmp分区，‘dfs -count’查看文件数，1100个，运行加上group by的临时任务（spark.sql.shuflle.partitions=1），查看结果目录，文件数=1，成功。

总结：

1）方便的话，可以采用coalesce()方法和repartition()方法
2）如果任务逻辑简单，数据量少，可以直接降低并行度
3）任务逻辑复杂，数据量很大，原任务大并行度计算写到临时分区，再加两个任务：
一个用来将临时分区的文件用小并行度（加宽依赖）合并成少量文件到实际分区
另一个删除临时分区
4）hive任务减少小文件相对比较简单，可以直接设置参数，如：
Map-only的任务结束时合并小文件：

set hive.merge.mapfiles =true

在Map-Reduce的任务结束时合并小文件：

set hive.merge.mapredfiles=true

当输出文件的平均大小小于1GB时，启动一个独立的map-reduce任务进行文件merge：

set hive.merge.smallfiles.avgsize=1024000000