快捷導(dǎo)航

Spark?SQL小文件問題處理

更新時(shí)間：2023年04月07日 09:56:56 作者：寶哥大數(shù)據(jù)

大量的小文件會影響Hadoop集群管理或者Spark在處理數(shù)據(jù)時(shí)的穩(wěn)定性，這篇文章主要介紹了Spark?SQL小文件問題的處理，感興趣的同學(xué)可以借鑒一下

1.1、小文件危害

大量的小文件會影響Hadoop集群管理或者Spark在處理數(shù)據(jù)時(shí)的穩(wěn)定性：

1.Spark SQL寫Hive或者直接寫入HDFS，過多的小文件會對NameNode內(nèi)存管理等產(chǎn)生巨大的壓力，會影響整個(gè)集群的穩(wěn)定運(yùn)行
2.容易導(dǎo)致task數(shù)過多，如果超過參數(shù)spark.driver.maxResultSize的配置（默認(rèn)1g），會拋出類似如下的異常，影響任務(wù)的處理

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)

當(dāng)然可以通過調(diào)大spark.driver.maxResultSize的默認(rèn)配置來解決問題，但如果不能從源頭上解決小文件問題，以后還可能遇到類似的問題。此外，Spark在處理任務(wù)時(shí)，一個(gè)分區(qū)分配一個(gè)task進(jìn)行處理，多個(gè)分區(qū)并行處理，雖然并行處理能夠提高處理效率，但不是意味著task數(shù)越多越好。如果數(shù)據(jù)量不大，過多的task運(yùn)行反而會影響效率。最后，Spark中一個(gè)task處理一個(gè)分區(qū)從而也會影響最終生成的文件數(shù)。

1.2、產(chǎn)生小文件過多的原因

1、流式處理中，每個(gè)批次的處理執(zhí)行保存操作也會產(chǎn)生很多小文件
2、為了解決數(shù)據(jù)更新問題，同一份數(shù)據(jù)保存了不同的幾個(gè)狀態(tài)，也容易導(dǎo)致文件數(shù)過多

1.3、如何解決這種小文件的問題呢？

通過repartition或coalesce算子控制最后的DataSet的分區(qū)數(shù), 注意repartition和coalesce的區(qū)別
將Hive風(fēng)格的Coalesce and Repartition Hint 應(yīng)用到Spark SQL 需要注意這種方式對Spark的版本有要求，建議在Spark2.4.X及以上版本使用，

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

小文件定期合并可以定時(shí)通過異步的方式針對Hive分區(qū)表的每一個(gè)分區(qū)中的小文件進(jìn)行合并操作

上述只是給出3種常見的解決辦法，并且要結(jié)合實(shí)際用到的技術(shù)和場景去具體處理，比如對于HDFS小文件過多，也可以通過生成HAR 文件或者Sequence File來解決。

1.3.1、調(diào)優(yōu)參數(shù)

在小文件場景下，您可以通過如下配置手動指定每個(gè)Task的數(shù)據(jù)量（Split Size），確保不會產(chǎn)生過多的Task，提高性能。

當(dāng)SQL邏輯中不包含Shuffle操作時(shí)，設(shè)置此配置項(xiàng)，不會有明顯的性能提升。

參數(shù)	描述	默認(rèn)值
spark.sql.small.file.combine	用于設(shè)置是否開啟小文件優(yōu)化。 “true”表示開啟。開啟后，可以避免過多的小Task。	false
spark.sql.small.file.split.size	合并小文件后，用于指定單個(gè)Task期望的數(shù)據(jù)量。單位：Byte	256000000

set spark.default.parallelism = 400;

/*+ coalesce(40) */ 調(diào)整最后的task個(gè)數(shù)；

SELECT age, name FROM person DISTRIBUTE BY age;//按照某個(gè)字段重新分區(qū)重新分區(qū)。
對于使用動態(tài)分區(qū)的任務(wù)，使用distribute by。

insert overwrite table dm.dm_grw_retain_abtest_sd partition (year, month, day, retain_days)
select ……
distribute by retain_days -- 最終每個(gè)子分區(qū)一個(gè)文件
distribute by retain_days, cast(rand()*7 as int) -- 最終每個(gè)子分區(qū)7個(gè)文件

到此這篇關(guān)于Spark SQL小文件問題處理的文章就介紹到這了,更多相關(guān)SQL小文件問題處理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: