Spark SQL小文件问题处理，

Spark SQL小文件问题处理，

目录

• 1.1、小文件危害
• 1.2、产生小文件过多的原因
• 1.3、如何解决这种小文件的问题呢？
• 1.3.1、调优参数

1.1、小文件危害

大量的小文件会影响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中一个task处理一个分区从而也会影响最终生成的文件数。

1.2、产生小文件过多的原因

1、流式处理中，每个批次的处理执行保存操作也会产生很多小文件
2、为了解决数据更新问题，同一份数据保存了不同的几个状态，也容易导致文件数过多

1.3、如何解决这种小文件的问题呢？

1. 通过repartition或coalesce算子控制最后的DataSet的分区数, 注意repartition和coalesce的区别
2. 将Hive风格的Coalesce and Repartition Hint 应用到Spark SQL 需要注意这种方式对Spark的版本有要求，建议在Spark2.4.X及以上版本使用，

示例：
INSERT ... SELECT /*+ COALESCE(numPartitions) */ ...
INSERT ... SELECT /*+ REPARTITION(numPartitions) */ ...

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

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

1.3.1、调优参数

在小文件场景下，您可以通过如下配置手动指定每个Task的数据量（Split Size），确保不会产生过多的Task，提高性能。

当SQL逻辑中不包含Shuffle操作时，设置此配置项，不会有明显的性能提升。

参数	描述	默认值
spark.sql.small.file.combine	用于设置是否开启小文件优化。 “true”表示开启。开启后，可以避免过多的小Task。	false
spark.sql.small.file.split.size	合并小文件后，用于指定单个Task期望的数据量。 单位：Byte	256000000

set spark.default.parallelism = 400;

/*+ coalesce(40) */ 调整最后的task个数；

SELECT age, name FROM person DISTRIBUTE BY age;//按照某个字段重新分区重新分区。
对于使用动态分区的任务，使用distribute by。

insert overwrite table dm.dm_grw_retain_abtest_sd partition (year, month, day, retain_days)
select ……
distribute by retain_days -- 最终每个子分区一个文件
distribute by retain_days, cast(rand()*7 as int) -- 最终每个子分区7个文件

到此这篇关于Spark SQL小文件问题处理的文章就介绍到这了,更多相关SQL小文件问题处理内容请搜索PHP之友以前的文章或继续浏览下面的相关文章希望大家以后多多支持PHP之友！

您可能感兴趣的文章:

• Spark SQL的自定义函数UDF使用
• Spark JDBC操作MySQL方式详细讲解
• Spark SQL关于性能调优选项详解
• 详解Spark Sql在UDF中如何引用外部数据
• 大数据Spark Sql中日期转换FROM_UNIXTIME和UNIX_TIMESTAMP的使用
• springboot集成spark并使用spark-sql的示例详解
• 教你如何让spark sql写mysql的时候支持update操作