Spark SQL 小文件问题处理

在生产中,无论是通过SQL语句或者Scala/Java等代码的方式使用Spark SQL处理数据,在Spark SQL写数据时,往往会遇到生成的小文件过多的问题,而管理这些大量的小文件,是一件非常头疼的事情。

大量的小文件会影响Hadoop集群管理或者Spark在处理数据时的稳定性:

1. Spark SQL写Hive或者直接写入HDFS,过多的小文件会对NameNode内存管理等产生巨大的压力,会影响整个集群的稳定运行

2. 容易导致task数过多,如果超过参数spark.driver.maxResultSize的配置(默认1g),会抛出类似如下的异常,影响任务的处理

Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 478 tasks (2026.0 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)

当然可以通过调大spark.driver.maxResultSize的默认配置来解决问题,但如果不能从源头上解决小文件问题,以后还可能遇到类似的问题。

此外,Spark在处理任务时,一个分区分配一个task进行处理,多个分区并行处理,虽然并行处理能够提高处理效率,但不是意味着task数越多越好。如果数据量不大,过多的task运行反而会影响效率。

下面通过一个例子,Spark SQL写数据时,导致产生分区数"剧增"的典型场景,通过分区数"剧增",以及Spark中task数和分区数的关系等,来倒推小文件过多的可能原因(这里的分区数是指生成的DataSet/RDD的分区数,不是Hive分区表的分区概念):

1. 现象

1) 对表test_tab进行写入操作
2) t1的分区数是100,t2的分区数是200,union all后生成的tmp分区数是300
3) test_tab产生的小文件数基本也在300左右
select * from t1 union all select * from t2 as tmp;insert overwrite table test_tab select * from tmp;

2. 分析

1)执行上述insert操作时的分区并行度,主要受tmp的分区数(对应一个DataSet)影响,

2)tmp的分区数主要受t1、t2以及union all的影响

3)暂且不考虑t1或t2是物理表还是经过其他处理生成的临时表,它们的分区数是确定的,这里主要看经过union all处理后,生成的tmp的分区数和t1、t2的分区数有何关系?

4)Spark SQL语句中的union all对应到DataSet中即为unionAll算子,底层调用union算子

在之前的文章《重要|Spark分区并行度决定机制》中已经对Spark RDD中的union算子对union产生的新的RDD的分区数是如何受被union的多个RDD的影响的,做过详细介绍,这里直接给出结论:

Spark SQL 小文件问题处理

同样的这种机制也可以套用到Spark SQL中的DataSet上,那么就很好解释了tmp的分区数为什么等于t1和t2的分区数的和。

最后,Spark中一个task处理一个分区从而也会影响最终生成的文件数。

当然上述只是以Spark SQL中的一个场景阐述了小文件产生过多的原因之一(分区数过多)。在数仓建设中,产生小文件过多的原因有很多种,比如:

1. 流式处理中,每个批次的处理执行保存操作也会产生很多小文件

2. 为了解决数据更新问题,同一份数据保存了不同的几个状态,也容易导致文件数过多

那么如何解决这种小文件的问题呢?

1. 通过repartition或coalesce算子控制最后的DataSet的分区数

注意repartition和coalesce的区别,具体可以参考文章《重要|Spark分区并行度决定机制》

2. 将Hive风格的Coalesce and Repartition Hint 应用到Spark SQL需要注意这种方式对Spark的版本有要求,建议在Spark2.4.X及以上版本使用,示例:

INSERT ... SELECT /*+ COALESCE(numPartitions) */ ...
INSERT ... SELECT /*+ REPARTITION(numPartitions) */ ...

3. 小文件定期合并

可以定时通过异步的方式针对Hive分区表的每一个分区中的小文件进行合并操作

上述只是给出3种常见的解决办法,并且要结合实际用到的技术和场景去具体处理,比如对于HDFS小文件过多,也可以通过生成HAR 文件或者Sequence File来解决。

推荐文章:

Spark SQL | 目前Spark社区最活跃的组件之一

Spark存储Parquet数据到Hive,对map、array、struct字段类型的处理Spark SQL解析查询parquet格式Hive表获取分区字段和查询条件

关于HDFS应知应会的几个问题Spark RDD详解

Spark和Spring整合处理离线数据Spark流式状态管理


关注微信公众号:大数据学习与分享,获取更对技术干货

上一篇:TX Textcontrol 使用总结四——打印


下一篇:Myeclipse 2014 javascript 添加 jquery 代码提示