HDFS小文件过多问题治理指南

目 录

1 目的 1

2 小文件产生的典型场景 2

3 小文件问题的危害 3

4 治理小文件问题的方案 4

4.1 监控 4

4.2 小文件产生过程优化控制 4

4.2.1 JAVA MapReduce: 4

4.2.2 HIVE 5

4.2.3 TEZ 6

4.2.4 SCALA 6

4.3 历史小文件合并归档优化 9

5 备注 11

1 目的

计划对Hadoop集群中的小文件问题进行重点整治，梳理问题背景、影响，以及对应的解决方案，以指导相关工作顺利推进。

2 小文件产生的典型场景

 数据源本身就包含大量的小文件。

 动态分区插入数据，产生大量的小文件，从而导致map数量剧增。

 reduce数量越多，小文件也越多(reduce的个数和输出文件是对应的)。

3 小文件问题的危害

 从Hive的角度看，小文件会开很多map，一个map开一个JVM去执行，所以这些任务的初始化，启动，执行会浪费大量的资源，严重影响性能。

 在HDFS中，每个小文件对象约占150byte，如果小文件过多会占用大量内存。这样NameNode内存容量严重制约了集群的扩展。

 任务计算过程中过多的读写请求对NameNode RPC产生巨大并发压力，极端情况下会对NN RPC请求产生风暴，导致NameNode无法响应其它请求，进而触发切换。

4 治理小文件问题的方案

4.1 监控

使用tableau，配置监控，查看对应业务线的DB、Table的小文件分布情况：

4.2 小文件产生过程优化控制

 使用Sequencefile（orc、parquet）作为表存储格式，不要用textfile，在一定程度上可以减少小文件。

 减少reduce的数量(可以使用参数进行控制)。

 少用动态分区，用时记得按distribute by分区。（注：DISTRIBUTE BY能够控制map的输出在reduce中如何划分。其可以按照指定的字段对数据进行划分到不同的输出reduce/文件中。distribute by与group by对比都是按key值划分数据都使用reduce操作唯一不同的是distribute by只是单纯的分散数据，而group by把相同key的数据聚集到一起，后续必须是聚合操作。）

参考链接：

https://www.iteblog.com/archives/1533.html

4.2.1 JAVA MapReduce:

1) job.setInputFormatClass(CombineTextInputFormat.class)

//指定SplitSize---可选项

CombineTextInputFormat.setMaxInputSplitSize(job, 512*1024*1024L);

//指定输入路径 ---多路径输入情况（1）：

CombineTextInputFormat.addInputPath(job, new Path(args[0]));

CombineTextInputFormat.addInputPath(job, new Path(args[1]));

//指定输入路径,以及指定输入格式 ---多路径输入情况（2）：

MultipleInputs.addInputPath(job, new Path(args[0]), CombineTextInputFormat.class);

MultipleInputs.addInputPath(job, new Path(args[1]), CombineTextInputFormat.class);

2) 关键参数：

//每个Map最大输入大小(这个值决定了合并后文件的数量)

set mapreduce.input.fileinputformat.split.maxsize=536870912;

//一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)

set mapreduce.input.fileinputformat.split.minsize.per.node=536870912;

//一个交换机下split的至少的大小(这个值决定了多个交换机上的文件是否需要合并)

set mapreduce.input.fileinputformat.split.minsize.per.rack=536870912;

代码控制示例（注：Hadoop默认配置dfs.blocksize=134217728，代码控制分片，非特殊情况不能小于默认值）：

final Long MAP_SPLIT_SIZE = 512 * 1024 * 1024L;

job.setConfiguration().set("mapreduce.input.fileinputformat.split.maxsize",MAP_SPLIT_SIZE.toString());

相关参考链接，如下：

http://www.cnblogs.com/skyl/p/4761662.html

4.2.2 HIVE

关键参数优化，如下：

设置map输入合并小文件的相关参数：

//每个Map最小输入大小（一般情况下无需显式设置该参数）

set mapreduce.input.fileinputformat.split.minsize=240000000

//每个Map最大输入大小(这个值决定了合并后文件的数量)

set mapreduce.input.fileinputformat.split.maxsize=536870912;

//一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)

set mapreduce.input.fileinputformat.split.minsize.per.node=536870912;

//一个交换机下split的至少的大小(这个值决定了多个交换机上的文件是否需要合并)

set mapreduce.input.fileinputformat.split.minsize.per.rack=536870912;

//执行Map前进行小文件合并

set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

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

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

set hive.merge.mapfiles=true

//设置reduce端输出进行合并，默认为false。

set hive.merge.mapredfiles=true

//设置合并文件的大小 。

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

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

set hive.merge.smallfiles.avgsize=134217728

4.2.3 TEZ

//在Map-only的任务结束时合并小文件

set hive.merge.tezfiles=true

//TEZ任务分片最小限制

set tez.grouping.min-size=536870912

//TEZ任务分片最大限制

set tez.grouping.max-size= 1073741824

4.2.4 SCALA

1) 输入文件类型设置为 CombineTextInputFormat

val data = sc.newAPIHadoopFile(args(1),

classOf[CombineTextInputFormat],

classOf[LongWritable],

classOf[Text], hadoopConf)

.map { //TODO }

2) 根据具体情况将分区数设置为合适的值（注：对于小文件的控制及运行效率考虑，建议尽可能设置较小的值，默认情况下rdd根据dfs.blocksize=134217728（128M）进行分区）：

Spark RDD 分区算子：coalesce、repartition，相关用法链接，说明如下：

http://lxw1234.com/archives/2015/07/341.htm

https://blog.csdn.net/u011981433/article/details/50035851

http://www.cnblogs.com/fillPv/p/5392186.html

3) Spark sql:

val conf = new SparkConf()

val spark = SparkSession.builder().appName("HiveDemo").config(conf).enableHiveSupport().getOrCreate()

import spark.implicits._

spark.sql("set mapreduce.input.fileinputformat.split.maxsize=536870912")

spark.sql("set mapreduce.input.fileinputformat.split.minsize.per.node=536870912")

spark.sql("set mapreduce.input.fileinputformat.split.minsize.per.rack=536870912")

spark.sql("set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat")

spark.sql("set hive.merge.mapfiles=true")

spark.sql("set hive.merge.mapredfiles=true")

spark.sql("set hive.merge.size.per.task=536870912")

spark.sql("set hive.merge.smallfiles.avgsize=134217728")

spark.stop()

根据实际情况指定：

spark.sql.shuffle.partitions的值（默认值：200）

spark.sql.files.maxPartitionBytes=536870912（默认：134217728 (128 MB)）

spark sql自适应计划（这块儿要跟进）：

https://www.iteblog.com/archives/2319.html

spark sql自适应patch（对dataset api和sql部分）：

https://github.com/apache/spark/pull/20303

Spark SQL 2.0版本之前，reduce的个数是通过spark.default.parallelism和spark.sql.shuffle.partitions两个参数进行配置。如果配置过大，将会导致下游产生很多碎片化的Task，或者下游HDFS产生很多小文件。如果设置过小，将会导致单个ReduceTask计算负载过大：

1

2 、Spark SQL 2.0+支持通过spark.sql.adaptive.enabled来设置reduce大小自适应

、spark.sql.files.ignoreCorruptFiles

hive.exec.reducers.bytes.per.reducer=256 * 1000 * 1000

=>（匹配的spark sql参数，如下：）

spark.sql.adaptive.shuffle.targetPostShuffleInputSize=1073741824（默认值：64 * 1024 * 1024）

mapred.reduce.tasks（mapreduce.job.reduces）

=> （匹配的spark sql参数，如下：）

spark.sql.shuffle.partitions

hive.mapjoin.smalltable.filesize=25000000

=>（匹配的spark sql参数，如下：）

spark.sql.autoBroadcastJoinThreshold=26214400 (默认值：10485760，10L * 1024 * 1024,-1禁用)

transformation

输入小文件合并的改进 （增加支持自定义inputFormat类）

默认采用建表时指定的InpurFormat，如果是默认的TextInputFormat，当小文件比较多是可能会导致RDD的partition数太多，导致性能下降。

解决办法：

通过参数允许用户指定InputFormat，在TableReader中反射生成对应的InputFormat对象并传入到HadoopRDD的构造函数中。

使用方法：

set spark.sql.hive.inputformat=org.apache.hadoop.mapred.lib.CombineTextInputFormat;

输出小文件合并的改进 （增加自动合并结果文件）

当spark.sql.shuffle.partitions设置的比较大且结果数据集比较小时，会产生大量的小文件（文件数等同spark.sql.shuffle.partitions）。

解决办法：

在最后的执行计划中加入一个repartition transformation。通过参数控制最终的partitions数且不影响shuffle partition的数量。

使用方法：

set spark.sql.result.partitions=10;

• 支持yarn-cluster模式，减小client的负载默认的yarn-client模式下Scheduler会运行在client上，加重client机器的负载。

解决办法：

让sparkSQL工具支持yarn-cluster模式。

1）在Yarn集群上部署SparkSQL依赖的hive metastore jar包。

2）开通Yarn nodemanager节点访问Hive metastore数据库的权限。

3）解决“\”“转义问题。如 spark-hive –e “select * fromuser where name = \”张三””;在yarn-cluster模式中会触发两次command执行从而导致“\”被转义两次。

使用示例：

/opt/spark/bin/spark-submit --class org.trainning.spark.sql.hive.SparkSQLCLIDriver \

--master yarn \

--deploy-mode cluster \

--conf spark.dynamicAllocation.enabled=true \

--conf spark.shuffle.service.enabled=true \

--conf spark.dynamicAllocation.minExecutors=1 \

--conf spark.dynamicAllocation.maxExecutors=1000 \

--conf spark.sql.adaptive.enabled=true \

--driver-memory 3g \

--executor-memory 3g \

--name test-sparksql \

--queue default \

hdfs:///apps/hive/udf/SparkSQLCLIDriver.jar -e "select count(*) from saledata.tbldate"

4) groupByKey、reduceByKey、aggregatebykey、combinebykey根据场景区别使用：

注：尽量避免使用groupByKey，reduceByKey和groupByKey都会产生相同的答案，但reduceByKey示例在大型数据集上运行得更好。 这是因为Spark知道它可以在混洗数据之前将输出与每个分区上的公用密钥相结合。

相关参考链接，如下：

https://www.jianshu.com/p/0c6705724cff

https://stackoverflow.com/questions/43364432/spark-difference-between-reducebykey-vs-groupbykey-vs-aggregatebykey-vs-combineb

4.3 历史小文件合并归档优化

 使用hadoop archive命令把小文件进行归档。

参考链接：

https://hadoop.apache.org/docs/current/hadoop-archives/HadoopArchives.html

https://blog.csdn.net/lastsweetop/article/details/9123155

 对于历史的小文件表处理，参考如下（注:hive merge/split等相关参数在生产集群的控制平台都已调整，无需特殊处理）：

例：

set hive.exec.dynamic.partition = true; --动态分区必设置参数

set hive.exec.dynamic.partition.mode=nonstrict; ---动态分区必设置参数

set hive.optimize.sort.dynamic.partition=false; ---分区内全局排序，默认是关闭，如果2级以上动态分区则建议开启。

set hive.exec.max.dynamic.partitions.pernode=20000;---根据具体情况来设置

set hive.exec.max.dynamic.partitions=20000; ---根据具体情况来设置

insert overwrite table tab_a partition(key) select a,……,key from tab_a

或者脚本处理：

small-file.sh（非分区表或非algo库的需改写脚本）

#!/bin/bash

srcTable=$1

tmpTable="${srcTable}_new"

cat <<EOF >.small.hql

use algo;

create table if not exists ${tmpTable} like ${srcTable};

set hive.exec.dynamic.partition.mode=nonstrict;

set hive.exec.max.dynamic.partitions.pernode=100000;

set hive.exec.max.dynamic.partitions=100000;

set hive.exec.max.created.files=100000;

insert into ${tmpTable} partition(stat_date) select * from ${srcTable};

drop table ${srcTable};

ALTER TABLE ${tmpTable} RENAME TO ${srcTable};

EOF

5 备注