PySpark合并多分区Parquet文件的性能调优问题

问题背景 #

处理340个日期分区Parquet文件（共13000个文件、约7GB数据）时，采用循环逐个读取分区→Schema对齐→循环Union的方式耗时超7小时，需从代码层面优化，并解答以下疑问：

疑问1：循环逐个读取分区是否为串行操作？能否实现并行读取？ #

是，默认循环逐个读取属于Driver端串行操作——每个spark.read.parquet()调用都是在Driver进程中依次触发，无法利用Spark的分布式并行能力，这是性能瓶颈的核心原因之一。

可以实现并行读取，推荐两种方案：

1. 直接读取父目录（优先选择）
   若分区路径符合标准Hive分区格式（/feed=abc/date=YYYYMMDD），直接读取父目录让Spark自动并行扫描所有分区：

   val rawDf = spark.read.option("mergeSchema", "true").parquet("/feed=abc")
   

   Spark会自动识别分区并并行处理，无需手动遍历路径。之后再对rawDf统一执行Schema对齐逻辑，一次性转换到目标Schema。

2. 分布式并行处理路径列表
   若必须手动遍历路径，可将路径列表转为Spark Dataset，通过mapPartitions实现分布式并行读取与Schema对齐：

   import org.apache.spark.sql.SparkSession
   import org.apache.spark.sql.types.StructType
   
   val targetSchema: StructType = // 你的目标Schema定义
   val partitionPaths: List[String] = // 收集到的所有分区路径
   
   // 并行读取并对齐Schema
   val processedDfs = spark.createDataset(partitionPaths)
     .mapPartitions { pathIter =>
       val spark = SparkSession.getActiveSession.get
       pathIter.map { path =>
         val df = spark.read.option("mergeSchema", "true").parquet(path)
         // 调用自定义的Schema对齐函数
         alignToTargetSchema(df, targetSchema)
       }
     }
   
   // 一次性合并所有处理后的DataFrame
   val finalDf = processedDfs.reduce(_ unionByName _)
   

   注：之前尝试reduce失败，大概率是因为未确保所有DataFrame的Schema完全对齐，或使用了union（依赖列顺序）而非unionByName（按列名匹配）。

疑问2：每次Union后缓存final_df无性能提升，是否应该缓存？ #

不建议每次Union后缓存，原因如下：

1. 循环Union会导致DataFrame的血缘（Lineage）无限膨胀——每次Union都会将之前所有的依赖关系加入执行计划，即使缓存旧的final_df，新生成的final_df仍会依赖旧缓存和新读取的DataFrame，无法缩短血缘。
2. 频繁缓存合并会额外占用内存资源，且缓存合并的IO开销会抵消并行处理的收益。

正确的缓存时机：在所有分区处理完成、一次性合并得到最终finalDf后，若后续还有多次操作，再执行finalDf.cache()。

额外代码调优建议 #

1. 优化Schema对齐逻辑
   编写高效的Schema对齐函数，避免冗余操作：

   • 对Struct类型子列，直接用struct函数逐个转换子列类型，避免全表扫描；
   • 用selectExpr批量指定列转换逻辑，减少多次withColumn的开销。
     示例：

   def alignToTargetSchema(df: DataFrame, targetSchema: StructType): DataFrame = {
     val selectExpr = targetSchema.fields.map { field =>
       field.dataType match {
         case structType: StructType =>
           // 处理Struct类型子列转换
           val structExpr = structType.fields.map(subField =>
             s"cast(${field.name}.${subField.name} as ${subField.dataType.sql}) as ${subField.name}"
           ).mkString(",")
           s"struct($structExpr) as ${field.name}"
         case _ =>
           s"cast(${field.name} as ${field.dataType.sql}) as ${field.name}"
       }
     }
     df.selectExpr(selectExpr: _*)
   }
   

2. 调整Spark读取参数

   • 全局开启Parquet Schema合并：spark.sql.parquet.mergeSchema=true，避免每个读取操作重复设置；
   • 调整文件分区大小：spark.sql.files.maxPartitionBytes=67108864（64MB），让Spark生成更合理的Task数量，提升并行度。

3. 避免小文件问题
   若合并后需输出，可通过repartition或coalesce减少输出文件数量，避免后续操作的小文件开销。

内容的提问来源于stack exchange，提问作者Kaushik Ghosh