Spark 处理小文件

2022-03-27 19:51:21 浏览数 (1)

1. 小文件合并综述

1.1 小文件表现

不论是Hive还是Spark SQL在使用过程中都可能会遇到小文件过多的问题。小文件过多最直接的表现是任务执行时间长,查看Spark log会发现大量的数据移动的日志。我们可以查看log中展现的日志信息,去对应的路径下查看文件的大小和个数。

代码语言:javascript复制
hdfs dfs -count <path>

通过上述命令可以查看文件的个数以及大小。count查看出的文件大小单位是B,需要转换为MB。

在spark官方的推荐文档中,parquet格式的文件推荐大小是128MB,小于该大小的均可以称之为小文件,在实际的工作,往往小文件的大小仅仅为几KB,表现为,可能文件大小为几百MB,但是文件个数可能到达了几十万个。一般来说,我们可以通过简单相除获得文件的平均大小,如果文件数目不多,我们也可以通过下述命令获得每个文件的大小。

代码语言:javascript复制
hdfs dfs -du <path>

1.2 小文件的危害

1.任务执行时间长

2.真实的文件大小独占一个数据存储块,存放到DataNode节点中。同时 DataNode一般默认存三份副本,以保障数据安全。同时该文件所存放的位置也写入到NameNode的内存中,如果有Secondary NameNode高可用节点,也可同时复制一份过去。NameNode的内存数据将会存放到硬盘中,如果HDFS发生重启,将产生较长时间的元数据从硬盘读到内存的过程。

3.不论在Hive还是在Spark中,每一个存储块都对应一个Map程序,一个Map呈现就需要一个JVM,启动一个JVM去读取或者写小文件是吃力不讨好的行为。在实际的生产中,为了更好的管理集群资源,一般会要求程序执行时限制Executor数量和每个Executor的核心数量,需要频繁创建Executor来读取写入。

  1. 其元数据会占用大量 namenode内存(一个元数据大概150字节),影响namenode性能

5.影响磁盘寻址时间

1.3 小文件出现的原因

  1. 启用了动态分区,往动态分区表插入数据时,会插入大量小文件
  2. reduce的数量设置的较多,到reduce处理时,会分配到不同的reduce中,会产生大量的小文件
  3. 源数据文件就存在大量的小文件

1.4 小文件合并的通俗理解

小文件合并,本质上就是通过某种操作,将一系列小文件合并成大文件。我们知道,以MapReduce为代表的大数据系统,都习惯用K-V键值对的形式来处理文件,最后文件落盘,也是一个reduce对应一个输出文件。所以直观上,我们可以减少reduce数量,达到减少文件数量的目的。

从Map到Reduce需要一个Shuffle过程,所以我们将小文件合并理解为通过一个Shuffle,合并小文件成一个大文件。基于这样的思想,我们的策略可以分为两类:一类是原来的计算已经有Shuffle了,那么我们可以认为控制输出文件的数量;二类是强制触发Shuffle,进行小文件合并。

2. 小文件合并方法策略

1-设置参数(一般用于Hive)

代码语言:javascript复制
-- 每个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;
 -- 每个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;
 -- 设置map端输出进行合并,默认为true
 set hive.merge.mapfiles = true
 -- 设置reduce端输出进行合并,默认为false
 set hive.merge.mapredfiles = true
 -- 设置合并文件的大小
 set hive.merge.size.per.task = 256*1000*1000
 -- 当输出文件的平均大小小于该值时,启动一个独立的MapReduce任务进行文件merge。
 set hive.merge.smallfiles.avgsize=16000000

2-distribute by rand()

往动态分区插入数据时,在已经写好的SQL末尾加上distribute by rand()

代码语言:javascript复制
SET hive.exec.dynamic.partition.mode=nonstrict;
 SET hive.mapred.mode=nonstrict;
 INSERT OVERWRITE TABLE hive_demo.demo_table PARTITION (dt)
 SELECT *
 FROM hive_demo.demo_table
 DISTRIBUTE BY RAND() ;

该算子只是起到打散的效果,但是我们还要设置文件的大小,以免打散后仍然有小文件。

代码语言:javascript复制
set hive.exec.reducers.bytes.per.reducer=5120000000; -- 默认是1G,设置为5G

表示每个reduce的大小,Hive可以数据总量,得到reduce个数,假设hive认为会有10个reduce,那么,这里rand()则会为 x % 10

3-group by

我们知道,group by算子会触发Shuffle,因此只要我们设置好Shuffle时的文件个数就好,在Spark SQL中,我们可以设置partition个数,因为一个partition会对应一个文件。

代码语言:javascript复制
select name, count(1)
 from new_table
 group by name
 ;

上述的操作,会触发shuffle,因此我们再设置partition个数。

代码语言:javascript复制
set spark.sql.shuffle.partition=10

则表示,shuffle后,只会产生10个partition.

4-repartition()

代码语言:javascript复制
select /*  repartition(10) */
        *
 from table;

5-coalesce()

代码语言:javascript复制
select /*  coalesce(10) */
        *
 from table; 

需要注意的是,4和5都是spark 2.4以及以后才会支持的。

0 人点赞