Kyuubi项目中Spark小文件合并问题分析与解决方案

问题背景

在Apache Kyuubi 1.8.0版本中，用户报告了一个关于Spark小文件合并的问题。从用户提供的截图来看，系统生成了大量小文件，但预期的文件合并功能并未触发。这种情况在数据湖或大数据处理场景中较为常见，会导致HDFS Namenode压力增大、查询性能下降等一系列问题。

问题分析

Spark SQL在写入数据时，默认会根据分区数和执行计划生成相应数量的文件。当数据量不大但分区数较多时，就容易产生小文件问题。Spark提供了多种机制来优化这种情况：

1. 自适应查询执行(AQE)：Spark 3.0引入的重要特性，可以动态调整执行计划
2. 分区合并：通过spark.sql.adaptive.coalescePartitions相关参数控制
3. 并行度优先策略：由spark.sql.adaptive.coalescePartitions.parallelismFirst参数控制

从技术专家的回复来看，问题的根源可能在于并行度优先策略的配置。当spark.sql.adaptive.coalescePartitions.parallelismFirst设置为true时，Spark会优先保持并行度，这可能导致即使存在小文件也不会进行合并。

解决方案

针对这个问题，可以采取以下解决方案：

1. 关闭并行度优先策略： 将spark.sql.adaptive.coalescePartitions.parallelismFirst设置为false，让Spark更积极地合并小文件

2. 调整合并分区相关参数：

   • spark.sql.adaptive.advisoryPartitionSizeInBytes：设置目标分区大小
   • spark.sql.adaptive.coalescePartitions.minPartitionNum：设置合并后的最小分区数
   • spark.sql.adaptive.coalescePartitions.initialPartitionNum：设置初始分区数

3. 针对特定作业的优化： 对于已知会产生小文件的作业，可以在SQL中显式指定repartition或coalesce操作

深入理解

Spark的文件合并机制实际上是在shuffle write阶段进行的优化。当启用AQE后，Spark会根据实际数据量动态调整reduce任务的数量。parallelismFirst参数控制了这个调整过程的策略：

• 当为true时，保持较高的并行度，适合计算密集型任务
• 当为false时，优先合并小分区，适合IO密集型任务

在Kyuubi这类SQL服务场景下，通常IO性能更为关键，因此关闭并行度优先策略是合理的建议。

最佳实践

除了上述解决方案外，还可以考虑以下最佳实践：

1. 监控小文件情况：定期检查HDFS目录，统计小文件数量和分布
2. 分层设置参数：根据作业类型设置不同的合并策略
3. 考虑使用Delta Lake/Iceberg：这些表格式内置了小文件合并机制
4. 定期执行压缩操作：对于历史数据，可以安排定期压缩作业

总结

Kyuubi作为Spark SQL服务，其小文件问题本质上是Spark执行计划的优化问题。通过合理配置AQE相关参数，特别是调整并行度优先策略，可以有效解决小文件合并不触发的问题。在实际生产环境中，建议根据数据特点和作业类型进行细致的参数调优，以达到最佳的性能和资源利用率平衡。