[大数据]大数据架构师全套课程之HDFS小文件

一．什么是 hdfs 小文件

小文件是指文件大小明显小于 HDFS 上块（block）大小（默认 64MB，在 Hadoop2.x 中默认为 128MB）的文件。如果一个文件的大小不是块大小的 30%， 那么它就是一个小文件。128X30%=38.4M

二．小文件带来的影响

1. 在 HDFS 中，任何 block，文件或者目录在内存中均以对象的形式存储， 每个对象约占 150byte，文件的元数据存储在 namenode 中，小文件过多，极大的占用 namonode 的内存，制约集群的扩展，节点压力增大。

2. 在 HDFS 中，MapReduce 是以进程运行的，在对小文件进行处理的时候，一个小文件对应一个 maptask，一个 maptask 会开启一个 JVM 进程，JVM 处理一个 maptask 后会关闭，这样 JVM 开关的时间会比处理 maptask的时间更长，耗费大量的时间在启动和关闭上。因为进程的开启销毁会严重性能，严重浪费了资源。

3. 在 HDFS 中，因为每次都需要从 NameNode 获取元信息，并且对应的DataNode 建立连接，如果访问大量小文件，还需要从一个 datanode 跳转到另外一个 datanode，大大降低了读取性能。

三．小文件产生的原因

1. 数据源有问题。使用 flume 拉取数据到 hdfs 上，落盘参数设置的不合理，出现许多小文件。

解决：

a2.sinks.k1.hdfs.rollInterval = 300

落盘时间设置太小 30 分钟

a2.sinks.k1.hdfs.rollSize = 104857600

落盘大小设置太小 128M

2. 数据本身的特点。在 HDFS 上存储大量的图片、短视频、短音频等文件。

3. hive 有问题。往动态分区插入数据，会产生很多小文件。进行 mapreduce计算。

4. 使用 Spark Streaming 从外部数据源接收数据，然后经过 ETL 处理之后存储到 HDFS 中，这种情况下在每个 Job 中会产生大量的小文件。

（1）小文件是因为 sparkstreaming 的微批处理模式和 DStream(RDD)的分布式(partition) 特性导致的

sparkstreaming 为每个 partition 启动一个独立的线程来处理数据，一旦文件输出到 HDFS，那么这个文件流就关闭了，再来一个 batch 的 parttition 任务，就再使用一个新的文件流。 一个 batch 为 10s，每个输出的 DStream 有 32 个 partition，那么一个小时产生的文件数将会达到

(3600/10) * 32=11520 个之多

（2）小文件的基数是：batch_number * partition_number

四．小文件解决的方法

1. sparkStreaming 解决

(1) 增加 batch 时间大小=减小 batch 个数。但是延迟会很大

(2) 减少 partition 个数。但是 32个线程在写256M 数据，现在可能变成了4个线程在写 256M数据,batch处理延迟增大

(3) 在 sparkstreaming 外再启动定时的批处理任务来合并 sparkstreaming 产生的小文件。但是注意时间划分，避免正在写入的文件

(4) sparkstreaming 提供的 foreach 这个 outout 类 api，调用 foreach 去 append，每个 batch 在写文件，把之前文件打开，追加。

但是判断一个文件已经达到某一个阈值时，就要产生一个新的文件进行追加了。

2. hdfs 自带的小文件解决

(1) Hadoop Archive（HRA—文件归档），一个高效地将小文件放入 HDFS 块中的文 件存档工具，它能够将多个小文件打包成一个 HAR 文件。

但是源文件目录以及源文件都不会自动删除需要手动删除； 一旦创建便不可修改，要想从中删除或者增加文件，必须重新建立存档文件；

(2) Sequence File，由一系列的二进制 key/value 组成，如果为 key 小文件名，value 为文件内容，则可以将大批小文件合并成一个大文件 。

(3) CombineInputFormat，将一个目录（可能包括多个小文件，不包括子目录）合 并到一个 InputSplit 中，作为一个 map 的输入 。

3. hive 解决

(1) 针对按分区插入数据， 可以使用 DISTRIBUTE BY rand() 将数据随机分配给Reduce，这样可以使得每个 Reduce 处理的数据大体一致. 设置每个 reducer 处理的大小为 5 个 G（默认 1G）

set hive.exec.reducers.bytes.per.reducer=5120000000;
设置 reduce 个数

set mapreduce.job.reduces=10
insert overwrite table A partition(dt)
select * from B
DISTRIBUTE BY cast(rand()*100 as int);
解释：如设置 reduce 数量为 10，则使用 rand()， 随机生成一个数 x % 10 ，

(2) 设置 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;

(3) 设置 map 输出和 reduce 输出进行合并的相关参数：

//设置 map 端输出进行合并，默认为 true

set hive.merge.mapfiles = true

//设置 reduce 端输出进行合并，默认为 falseset hive.merge.mapredfiles = true

//设置合并文件的大小

set hive.merge.size.per.task = 25610001000

//当输出文件的平均大小小于该值时，启动一个独立的 MapReduce 任务进行文件 merge。

set hive.merge.smallfiles.avgsize=16000000

(4) 开启 jvm 重用，前提条件 task 必需属于同一个 job

set mapred.job.reuse.jvm.num.tasks=20;

表示属于同一 job 的排队按顺序执行的 task 可以共享一个 JVM，也就是说第二轮的 map 可以重用前一轮的 JVM，而不是第一轮结束后关闭 JVM，第二轮再启动新的 JVM。

如果设置成-1，那么只要是同一个 job 的 task（无所谓多少个），都可以按顺序在一个 JVM 上连续执行。

如果 task 属于不同的 job，那么 JVM 重用机制无效，不同 job 的 task 需要不同的 JVM 来运行。

4. 在本地合并小文件后上传

HDFS 的 appendToFile 命令可以实现多个本地文件合并上传 HDFS。

hdfs dfs -appendToFile world1.txt world.txt /tmp/world.txt

5. 底层处理方案

(1) HDFS-8998:小文件社区改进 HDFS-8998，HDFS 自动启动一个服务，将小文件合并成大文件。DataNode 划分小文件区，专门存储小文件。一个 block 块满了开始使用下一个 block。

(2) HDFS-8286：将元数据从 namenode 从内存移到第三方 k-v 存储系统中。

(3) HDFS-7240：Apache Hadoop Ozone,hadoop 子项目，为扩展 hdfs 而生。

如需java大数据架构师大数据开发全套系列教程及学习资料请点击获取。

Hadoop3.X，spark3.X，Flink1.1x,Kylin,ClickHouse,Scala,Flume,实时数仓，CDH，机器学习，推荐系统，电商日志分析，智慧物流，智慧交通大数据分析，用户画像等项目实战教程全都有。