Spark Hive 小文件合并

您所在的位置：网站首页 › hive表分区的作用 › Spark Hive 小文件合并

Spark Hive 小文件合并

2023-12-11 16:00| 来源: 网络整理| 查看: 265

背景小文件带来的问题对于HDFS

从 NN RPC请求角度，文件数越多，读写文件时，对于NN的RPC请求就越多，增大NN压力。

从 NN 元数据存储角度，文件数越多，NN存储的元数据就越大。

对于下游流程

下游流程，不论是MR、Hive还是Spark，在划分分片（getSplits）的时候，都要从NN获取文件信息。这个过程的耗时与文件数成正比，同时受NN压力的影响。在NN压力大，上游小文件多的情况下，下游的getSplits操作就会比较慢。

作业生成的文件数

为了简化问题，假设：

不考虑一个task写出文件大小的限制，那么一个task对于一个分区（一个目录）只写出一个文件

没有数据的task不会写出文件

在MR、Spark中，设写出到HDFS的stage中task的个数为T

如果结果表没有分区，或者写出静态分区，则每个Task写出一个文件，那么最多会写出T个文件。

如果结果表有动态分区，不同的分区是写到不同的目录下，令第i个动态分区dp的基数（cardinality）为card(dpi)，那么如果k个动态分区，最多写出的文件数为card(dp1) * card(dp2) * ...* card(dpk ) * T。

如何减少作业生成的文件数

所以，控制最终输出的文件个数，可以从以下3个角度入手：

控制最终stage的task个数，也就是控制整个作业的并行度，具体来讲，可以从最开始单个map输入size，shuffle之后单个reduce的size两方面来控制。

在写入HDFS之后，计算平均文件大小，merge小文件(但是这种做法只能缓解NN元数据的压力，由于存在写小文件，统计平均文件大小，读小文件、写出大文件这一连串操作，会增加NN RPC的压力，在NN负载高的时候，还会增加作业本身的执行时间)。

控制最终stage的输入数据划分，让同一个分区的数据，尽量在一个task内。

Map端输入合并 Hadoop InputFormat关系

InputFormat

FileInputFormat

TextInputFormat

DeprecatedLzoTextInputFormat

OrcInputFormat

KeyValueTextInputFormat

SequenceFileInputFormat

CombineFileInputFormat

HiveInputFormat

ConbineHiveInputFormat

接口

具体实现

抽象类

桥接类

Spark getSplits路径

HadoopRDD -> wrapSplits -> InputFormat.getSplits

切分文件，得到原始splits，通过InputFormat调用具体的序列化实现，来后去splits

合并splits，根据1的结果，再结合spark.hadoopRDD.targetBytesInPartition参数的值，将splits合并

生成HadoopRDD

Hive getSplits路径

CombineHiveInputFormat(extends HiveInputFormat) -> CombineFileInputFormat -> FileInputFormat（InputFormat）

切分文件，得到原始splits，FileInputFormat，调用InputFormat获取splits

合并splits，CombineFileInputFormat，根据具体的InputFormat的实现划分的split结果，合并splits

CombineHiveInputFormat 根据输入的目录（可能多个），分别找到序列化方式，分别调用getSplits

每个类的作用 FileInputFormat

作用

可以将一个大的文件划分为一个或者多个split，但是不能合并，

原理

getSplits如何划分文件

在对一个文件，进行split切分的时候，computeSplitSize这个函数负责计算一个split的大小。

代码块

Java

protected long computeSplitSize(long goalSize, long minSize, long blockSize) { return Math.max(minSize, Math.min(goalSize, blockSize)); }

对于特定的文件，totalSize是文件大小，numSplits是一个文件期望的分片数，默认是1，所有goalSize默认值也就等于一个文件的大小。

代码块

Java

goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);

minSize 由参数 mapreduce.input.fileinputformat.split.minsize确定，默认值是1。

在默认情况下，最终的splitSize为：

splitSize

= Math.max(minSize, Math.min(goalSize, blockSize))

= Math.max(1, Math.min(fileSize, blockSize))

= Math.min(fileSize, blockSize)

因此，在默认情况下，一个split size为文件大小与blockSize的较小的一个，含义是：一个文件如果小于一个block就不分，如果大于一个block（可能分布在不同dn），就可以分割，看起来也是比较合理的。

但是如果设置了期望的分片数numSplits、最小的分片大小minSize，并且minSize > min(goalSize, blockSize)，minSize就会起作用，含义就是根据用户设置的最小分片大小minSize将文件切成多个splits。

新旧API的一点小差异

以上分析的是org.apache.hadoop.mapred.FileInputFormat里面的实现，DeprecatedLzoTextInputFormat就是依赖上面的实现来划分split

但是org.apache.hadoop.mapred这个包已经标记为deprecated，取而代之的是org.apache.hadoop.mapreduce下面的实现

对应的FileInputFormat为org.apache.hadoop.mapreduce.lib.input.FileInputFormat

org.apache.hadoop.mapreduce.lib.input.FileInputFormat与org.apache.hadoop.mapred.FileInputFormat划分split有一些变化，computeSplitSize的时候，公式为：

代码块

SQL

Math.max(minSize, Math.min(maxSize, blockSize))

minSize与blockSize不变，只是把goalSize换成了mapreduce.input.fileinputformat.split.maxsize（默认256M）。

CombineFileInputFormat

作用

将小的split合并成大的split

参数

含义

默认值

mapreduce.input.fileinputformat.split.maxsize

一个split的最大size

256000000 （256M）

mapred.min.split.size.per.node

一个node上的一个split的最小size

mapred.min.split.size.per.rack

一个rack上的一个split的最小size

这么多参数，如果我要在Hive中合并小文件，比如以256M为size合并，应该如何调参？

如果要想在map输入端以256M的size合并小文件，那么就把上面3个参数设置成256M就可以。

同时为了不必要的文件切分，把mapreduce.input.fileinputformat.split.minsize 也设置成256M。

CombineHiveInputFormat

在一个MR job中，可以读取不同InputFormat方式序列化的分区目录

Map端合并参数总结

如果想要在Map端按照大小S来合并文件，如何设置参数？

引擎

参数

值

Spark

spark.hadoopRDD.targetBytesInPartition

spark.hadoop.mapreduce.input.fileinputformat.split.maxsize

spark.hadoop.mapreduce.input.fileinputformat.split.minsize

Hive

mapreduce.input.fileinputformat.split.maxsize

mapreduce.input.fileinputformat.split.minsize

mapred.min.split.size.per.node

mapred.min.split.size.per.rack

以上分析适用于FileInputFormat（文本格式），对于ORC文件，还需要将hive.exec.orc.split.strategy 设置为ETL，原理可以参考Spark SQL参数调优指南

引擎

参数

值

Spark

spark.hadoopRDD.targetBytesInPartition

spark.hadoop.mapreduce.input.fileinputformat.split.maxsize

spark.hadoop.mapreduce.input.fileinputformat.split.minsize

spark.hadoop.hive.exec.orc.split.strategy

ETL

Hive

mapreduce.input.fileinputformat.split.maxsize

mapreduce.input.fileinputformat.split.minsize

mapred.min.split.size.per.node

mapred.min.split.size.per.rack

hive.exec.orc.split.strategy

ETL

Shuffle之后合并

无论是Spark还是Hive，在shuffle之后的合并都比较类似，都是根据上游的map的结果size，将多个map的结果合并给下游的一个reducer，具体的参数如下表：

功能

引擎

参数

值

确定下游合并的size

Spark

spark.sql.adaptive.shuffle.targetPostShuffleInputSize

134217728 （128M）

Hive

hive.exec.reducers.bytes.per.reducer

确定最大reducer个数

Spark

spark.sql.shuffle.partitions

2000

Hive

hive.exec.reducers.max

1009?

写入HDFS之后合并

原理都是统计目录下的平均文件大小，如果小于某个阈值，就再启动一个map job，来合并文件

Hive