继前三篇之后，我觉得Hadoop还有个很严重的小问题，需要特别列出来，所以有了本篇文章。
大数据之Hadoop从入门到放弃之hdfs大数据之Hadoop从入门到放弃之MapReduce大数据之Hadoop从入门到放弃之hadoop的小伙伴们

前言：一个关于"小"的大问题

如果你问我，Hadoop最让人头疼的问题是什么？

不是集群搭建，那有文档；不是参数调优，那有经验；不是数据倾斜，那有套路。

是小文件。

让我先讲一个真实的故事。

那是某大数据工程师入职新公司的第一周。老板说他说："小伙子，我们的Hadoop集群最近有点慢，你看看怎么回事。"

他信心满满地登上集群，敲下hdfs dfs -ls /user/hive/warehouse/，然后——

屏幕开始疯狂滚动......

滚了整整5分钟......

他数了一下，一个表的分区目录下，有47万个文件，平均每个文件3KB。

当时他的表情是这样的：😱

然后我看了一眼NameNode的内存使用：98%。

那个时候他终于明白了，为什么上一个大数据工程师离职了。

今天，让我们一起来彻底搞懂这个"小"问题。

一、什么是小文件问题？

1.1 小文件的定义

在HDFS的世界里，什么叫"小文件"？

简单来说：任何远小于HDFS Block大小（默认128MB）的文件，都可以被认为是"小文件"。

但真正可怕的是那些KB级别的文件——几KB、几十KB的文件，成千上万地堆在HDFS上，就像蚂蚁一样，单个不起眼，成群结队就能把大象（Hadoop）啃死。

1.2 小文件问题的本质

为什么小文件会成为问题？让我们从HDFS的设计说起。

核心问题：不管文件多大多小，每个文件在NameNode中都要占用约150字节的元数据空间。

让我们算一笔账：

这就是为什么我们说：HDFS天生不适合存储大量小文件。

二、小文件问题的影响

小文件不只是占用NameNode内存那么简单，它会带来一系列连锁反应。

2.1 影响全景图

2.2 对NameNode的影响

真实案例中的恐怖数字：

2.3 对MapReduce的影响

还记得MapReduce的原则吗？一个InputSplit对应一个Map Task。

Map Task的开销分析：

结论：处理小文件时，90%的时间都浪费在了启动和调度上，真正处理数据的时间只有10%。这就像你开车去买瓶酱油，路上花了1小时，买酱油只花了1分钟。

三、小文件是怎么产生的？

知己知彼，百战不殆。让我们看看小文件是怎么来的。

3.1 小文件的来源

3.2 典型场景分析

场景1：Flume采集日志

场景2：Hive动态分区

-- 危险操作！
INSERT OVERWRITE TABLE user_behavior
PARTITION (dt, hour, province)  -- 三级分区
SELECT * FROM tmp_data;

场景3：Spark任务输出

# 危险代码！
df.repartition(1000).write.parquet("/output/path")
# 产生1000个文件，如果数据量小，每个文件可能只有几MB

四、解决方案大全

好了，问题说清楚了，现在让我们来解决它！

4.1 解决方案全景图

4.2 方案一：HAR文件（Hadoop Archive）

HAR是Hadoop自带的归档工具，可以把多个小文件打包成一个归档文件。

HAR使用方法：

# 创建HAR归档
hadoop archive -archiveName myfiles.har -p /input/small_files /output/

# 查看HAR内容
hdfs dfs -ls har:///output/myfiles.har

# 读取HAR中的文件
hdfs dfs -cat har:///output/myfiles.har/file1.txt

# 在MapReduce/Hive中使用
# 直接用har://协议访问

HAR的优缺点：

4.3 方案二：SequenceFile

SequenceFile是Hadoop的一种二进制文件格式，可以把多个小文件合并成一个大文件。

合并小文件为SequenceFile的代码：

public class SmallFilesToSequenceFile {
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        
        // 输出SequenceFile
        Path outputPath = new Path("/output/merged.seq");
        SequenceFile.Writer writer = SequenceFile.createWriter(
            conf,
            SequenceFile.Writer.file(outputPath),
            SequenceFile.Writer.keyClass(Text.class),
            SequenceFile.Writer.valueClass(BytesWritable.class),
            SequenceFile.Writer.compression(CompressionType.BLOCK, new GzipCodec())
        );
        
        // 遍历小文件目录
        Path inputDir = new Path("/input/small_files");
        FileStatus[] files = fs.listStatus(inputDir);
        
        for (FileStatus file : files) {
            // Key: 文件名
            Text key = new Text(file.getPath().getName());
            
            // Value: 文件内容
            byte[] content = readFile(fs, file.getPath());
            BytesWritable value = new BytesWritable(content);
            
            writer.append(key, value);
        }
        
        writer.close();
        System.out.println("合并完成！" + files.length + " 个文件 -> 1 个SequenceFile");
    }
    
    private static byte[] readFile(FileSystem fs, Path path) throws IOException {
        FSDataInputStream in = fs.open(path);
        byte[] content = IOUtils.toByteArray(in);
        in.close();
        return content;
    }
}

SequenceFile的优缺点：

4.4 方案三：CombineFileInputFormat

这是MapReduce层面的解决方案，在读取时把多个小文件合并成一个InputSplit。

使用方法：

// 方法1：在代码中设置
job.setInputFormatClass(CombineTextInputFormat.class);
CombineTextInputFormat.setMaxInputSplitSize(job, 128 * 1024 * 1024); // 128MB
CombineTextInputFormat.setMinInputSplitSize(job, 64 * 1024 * 1024);  // 64MB

// 方法2：在配置中设置
conf.set("mapreduce.input.fileinputformat.split.maxsize", "134217728");
conf.set("mapreduce.input.fileinputformat.split.minsize", "67108864");

在Hive中使用：

-- 设置输入合并
SET hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
SET mapreduce.input.fileinputformat.split.maxsize=134217728;
SET mapreduce.input.fileinputformat.split.minsize=67108864;

4.5 方案四：Hive小文件处理

Hive提供了多种处理小文件的参数和方法。

4.5.1 Map端输入合并

-- 开启Map端输入合并
SET hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

-- 每个Map处理的最大数据量
SET mapreduce.input.fileinputformat.split.maxsize=256000000;

-- 每个节点处理的最小数据量
SET mapreduce.input.fileinputformat.split.minsize.per.node=128000000;

-- 每个机架处理的最小数据量
SET mapreduce.input.fileinputformat.split.minsize.per.rack=128000000;

4.5.2 输出端合并小文件

-- 开启输出合并
SET hive.merge.mapfiles=true;          -- Map-only任务结束时合并
SET hive.merge.mapredfiles=true;       -- MapReduce任务结束时合并
SET hive.merge.size.per.task=256000000; -- 合并后文件大小目标
SET hive.merge.smallfiles.avgsize=128000000; -- 小于此值触发合并

4.5.3 手动合并已有小文件

-- 方法1：INSERT OVERWRITE（会重写整个表/分区）
INSERT OVERWRITE TABLE your_table PARTITION(dt='2024-01-01')
SELECT * FROM your_table WHERE dt='2024-01-01';

-- 方法2：使用CONCATENATE（ORC表专用，不重写数据）
ALTER TABLE your_table PARTITION(dt='2024-01-01') CONCATENATE;

-- 方法3：使用distribute by控制输出文件数
INSERT OVERWRITE TABLE your_table PARTITION(dt='2024-01-01')
SELECT * FROM your_table WHERE dt='2024-01-01'
DISTRIBUTE BY rand();  -- 随机分布，控制Reducer数量

4.5.4 Hive小文件处理流程图

4.6 方案五：Spark小文件处理

4.6.1 控制输出文件数量

# 方法1：使用coalesce减少分区（窄依赖，效率高）
df.coalesce(10).write.parquet("/output/path")

# 方法2：使用repartition重新分区（会shuffle）
df.repartition(10).write.parquet("/output/path")

# 方法3：根据数据量动态计算分区数
data_size_mb = df.rdd.map(lambda x: len(str(x))).sum() / 1024 / 1024
num_partitions = max(1, int(data_size_mb / 128))  # 每个文件约128MB
df.coalesce(num_partitions).write.parquet("/output/path")

4.6.2 Spark 3.x 自适应执行(AQE)

# 开启自适应执行
spark.conf.set("spark.sql.adaptive.enabled", "true")

# 开启自动合并小分区
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# 合并后的目标大小
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

# 最小分区数
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionNum", "1")

4.6.3 读取时合并小文件

# 设置每个分区的最大字节数
spark.conf.set("spark.sql.files.maxPartitionBytes", "134217728")  # 128MB

# 设置打开文件的开销（用于计算是否合并）
spark.conf.set("spark.sql.files.openCostInBytes", "4194304")  # 4MB

# 小文件合并阈值
spark.conf.set("spark.sql.files.minPartitionNum", "1")

4.7 方案六：列式存储格式

使用ORC或Parquet等列式存储格式，天然支持高压缩比，同样数据量产生更少的文件。

转换为ORC格式：

-- 创建ORC表
CREATE TABLE user_behavior_orc (
    user_id STRING,
    action STRING,
    timestamp BIGINT
)
PARTITIONED BY (dt STRING)
STORED AS ORC
TBLPROPERTIES (
    "orc.compress"="SNAPPY",
    "orc.stripe.size"="67108864"  -- 64MB
);

-- 从原表导入数据
INSERT OVERWRITE TABLE user_behavior_orc PARTITION(dt)
SELECT user_id, action, timestamp, dt
FROM user_behavior_text;

4.8 方案七：源头治理

最好的解决方案是：从源头避免产生小文件。

Flume配置优化示例：

# 原来的配置（产生大量小文件）
agent.sinks.hdfs-sink.hdfs.rollInterval = 60
agent.sinks.hdfs-sink.hdfs.rollSize = 0
agent.sinks.hdfs-sink.hdfs.rollCount = 1000

# 优化后的配置
agent.sinks.hdfs-sink.hdfs.rollInterval = 3600      # 1小时滚动
agent.sinks.hdfs-sink.hdfs.rollSize = 134217728     # 128MB滚动
agent.sinks.hdfs-sink.hdfs.rollCount = 0            # 禁用按条数滚动
agent.sinks.hdfs-sink.hdfs.batchSize = 10000        # 批量写入

五、解决方案对比

5.1 各方案对比表

5.2 详细对比表

六、实战案例：一次完整的小文件治理

让我分享一个真实的小文件治理案例。

6.1 问题背景

6.2 治理方案

6.3 实施步骤

Step 1: 分析现状

# 统计文件数量
hdfs dfs -count /user/hive/warehouse/user_behavior | awk '{print $2}'
# 结果：47,382,156

# 统计平均文件大小
hdfs dfs -du -s /user/hive/warehouse/user_behavior
# 总大小：109GB，平均每个文件：2.3KB

# 查看分区分布
hdfs dfs -ls /user/hive/warehouse/user_behavior | head -20

Step 2: 创建新表

-- 新表：简化分区，使用ORC格式
CREATE TABLE user_behavior_optimized (
    user_id STRING,
    action STRING,
    hour INT,
    province STRING,
    timestamp BIGINT
)
PARTITIONED BY (dt STRING)
STORED AS ORC
TBLPROPERTIES (
    "orc.compress"="SNAPPY",
    "orc.stripe.size"="67108864"
);

Step 3: 数据迁移（带合并）

-- 设置合并参数
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.merge.mapredfiles=true;
SET hive.merge.size.per.task=268435456;  -- 256MB
SET hive.merge.smallfiles.avgsize=134217728;  -- 128MB

-- 迁移数据
INSERT OVERWRITE TABLE user_behavior_optimized PARTITION(dt)
SELECT 
    user_id,
    action,
    hour,
    province,
    timestamp,
    dt
FROM user_behavior;

Step 4: 验证效果

# 统计新表文件数量
hdfs dfs -count /user/hive/warehouse/user_behavior_optimized | awk '{print $2}'
# 结果：1,247

# 统计平均文件大小
hdfs dfs -du -s /user/hive/warehouse/user_behavior_optimized
# 总大小：31GB（ORC压缩后），平均每个文件：25MB

6.4 治理效果

七、最佳实践总结

7.1 黄金法则

1. 文件大小目标：单个文件128MB-1GB之间最佳
2. 文件数量控制：单个分区文件数不超过1000
3. 分区粒度：优先按天分区，避免按小时/分钟
4. 存储格式：优先使用ORC/Parquet
5. 定期巡检：每周检查文件数量，及时治理

7.2 监控脚本

#!/bin/bash
# 小文件监控脚本

THRESHOLD=10000  # 告警阈值
PATHS=(
    "/user/hive/warehouse/db1.db"
    "/user/hive/warehouse/db2.db"
)

for path in "${PATHS[@]}"; do
    echo "检查路径: $path"
    
    # 统计文件数量
    file_count=$(hdfs dfs -count $path | awk '{print $2}')
    
    # 统计小于1MB的文件数
    small_files=$(hdfs dfs -ls -R $path | awk '$5 < 1048576 {count++} END {print count}')
    
    echo "  总文件数: $file_count"
    echo "  小文件数(<1MB): $small_files"
    
    if [ "$small_files" -gt "$THRESHOLD" ]; then
        echo "  ⚠️ 警告：小文件数量超过阈值！"
        # 发送告警...
    fi
    echo ""
done

八、写在最后

小文件问题，是每个大数据工程师都会遇到的"成长之痛"。

它看起来不起眼——不就是文件小点嘛，能有什么问题？但当你的集群因为几千万个KB级别的小文件而濒临崩溃时，你就会明白：在大数据的世界里，"小"问题往往是"大"麻烦。

解决小文件问题的核心思想就一句话：

"合久必分，分久必合"——该拆的拆，该合的合，保持平衡。

记住几个关键数字：

• 文件大小：128MB-1GB
• 单分区文件数：< 1000
• NameNode内存占用：< 70%

如果你正在被小文件困扰，希望这篇文章能帮到你。如果你还没遇到这个问题——别担心，迟早会遇到的😄。

最后，送你一句话：

"治理小文件最好的时间是在它产生之前，其次是现在。"

祝你的集群永远健康！🐘

本文作者：一个曾经被4700万小文件支配的程序员

最惨经历：ls一个目录，等了20分钟还没出结果

系列番外完结，感谢阅读！🎉

附录：小文件问题面试题

1. 什么是HDFS小文件问题？为什么会产生这个问题？

   小文件指远小于Block大小的文件。问题根源是每个文件无论大小都要占用约150字节的NameNode内存，大量小文件会导致NameNode内存耗尽。

2. 小文件问题有哪些影响？

   NameNode内存压力、MapReduce效率下降（Task过多）、集群启动慢、查询性能差。

3. 如何解决小文件问题？

   事前：合理分区、控制输出文件数、采集端优化
   事中：CombineFileInputFormat、Hive/Spark合并参数
   事后：HAR归档、定期合并、转ORC/Parquet

4. Hive中如何处理小文件？

   输入端：使用CombineHiveInputFormat
   输出端：设置hive.merge.mapredfiles=true
   存量数据：ORC表用CONCATENATE，其他用INSERT OVERWRITE

5. CombineFileInputFormat的原理是什么？

   在生成InputSplit时，把多个小文件合并到一个Split中，从而减少Map Task数量。它是逻辑合并，不改变实际文件。

6. 为什么推荐使用ORC/Parquet格式？

   列式存储、高压缩比、支持谓词下推、ORC支持CONCATENATE快速合并。相同数据量，文件更少更小。