新增列后Parquet Spark count变慢及spark.sql.hive.filesourcePartitionFileCacheSize优化咨询

安全阿岳

2026-10-26

问题背景

场景说明：对旧Parquet数据集（存储路径path_of_old_parquet_file）完成字段转换、新增account_id列后，将全量转换结果写入新路径path_of_new_parquet_file，目标是让新路径的Spark count查询性能对齐旧数据集水平
运行环境栈：Spark 2.3.0.cloudera3，Scala 2.11.8，JDK 1.8.0_281（HotSpot 64-Bit Server VM）
存储规则：新旧数据集均按「日级日期、source（共3个取值）」做二级分区，总计约300个日期分区
存储规模：旧路径单副本总大小562.6G（三副本总计1.1T），新路径单副本总大小684.2G（三副本总计1.3T）

异常现象说明

旧数据集查询表现：执行spark.read.parquet("path_of_old_parquet_file").count，7分钟内返回结果，生成5604个分区，统计总数据量3450809070行
新数据集查询异常表现：完全相同的查询逻辑，耗时超过40分钟，生成分区数激增至89355，伴随两类固定告警：
- SharedInMemoryCache告警：提示受spark.sql.hive.filesourcePartitionFileCacheSize（当前配置262144000字节）大小约束，驱逐已缓存的表分区元数据，拖慢查询规划性能
- Utils告警：查询执行计划体积过大被截断
已验证无效的调整操作：
- 常规YARN资源参数调整
- 追加--conf spark.sql.shuffle.partitions=2000配置，无性能改善
当前固定资源配置：
- master yarn
- num-executors 50
- executor-cores 5
- driver-memory 50G
- executor-memory 14G
环境限制：暂无法访问Spark WebUI查看运行时指标，相关权限申请中

优化方案

核心根因定位：新数据集查询性能异常的本质是写数据时产生了海量小文件，导致Spark文件扫描阶段生成的输入分片数爆炸——由于count查询是纯文件扫描逻辑、无shuffle阶段，因此之前调整spark.sql.shuffle.partitions的操作完全不影响扫描分片数，无法解决问题。

第一步：定位小文件规模
- 执行hdfs dfs -count path_of_new_parquet_file/*/*，统计每个二级分区下的文件总数，对比新旧同分区路径的文件数差异，验证小文件问题的影响范围
第二步：从根源合并小文件
- 对新路径执行全量重写，按照旧数据集的文件大小标准合并小文件：旧数据集单文件对应约120M单副本大小、每个分片约61万行数据，可通过设置写入参数对齐规模：
```
// 设置单文件最大写入行数，对齐旧数据集的文件大小
spark.conf.set("spark.sql.files.maxRecordsPerFile", 600000)
val srcDF = spark.read.parquet("path_of_new_parquet_file")
srcDF.write
  .mode("overwrite")
  .partitionBy("date", "source")
  .parquet("path_of_new_parquet_file_merged")
```
- 验证合并后的路径文件数与旧路径基本一致后，将原新路径替换为合并后的路径
第三步：调整查询规划阶段配置，消除告警
- 调大分区元数据缓存大小：将spark.sql.hive.filesourcePartitionFileCacheSize从262144000字节（250M）调整为2684354560字节（2.5G），避免元数据缓存被提前驱逐，减少执行计划生成阶段的重复磁盘扫描
- 调大执行计划截断阈值：追加配置spark.sql.debug.maxToStringFields=1000，避免大体积执行计划被截断导致的规划阶段异常
第四步：适配Parquet扫描分片规则，对齐旧数据集的扫描效率
- 追加扫描参数：--conf spark.sql.files.maxPartitionBytes=134217728（128M），匹配HDFS块大小，让Spark读取Parquet时尽量按照HDFS块大小生成分片，避免将大量小文件拆分为无效分片
- 补充：如果暂时无法执行全量重写，可在查询时手动控制最大扫描分片数，绕过规划阶段的小文件拆分逻辑，直接执行下述语句即可快速拿到count结果：
```
spark.read.option("maxPartitions", 6000).parquet("path_of_new_parquet_file").count()
```

内容的提问来源于stack exchange，提问作者Abhishek Shah