高效去除S3上大量分区Parquet文件重复数据的Spark优化方案（Python/Scala）

看起来你在处理S3上海量分区Parquet文件的去重时遇到了性能瓶颈，之前用Pandas完全扛不住，现在的Spark基础方案也不够高效。我来给你一套针对性的优化方案，不管是Python还是Scala版本都能大幅提升处理速度。

核心优化思路 #

先说说为什么你的当前方案慢：

• 没有利用S3的分区过滤：你提到路径里包含Name/age，如果只处理指定的Name和Age组合，完全不用读所有分区的数据，这能直接砍掉大部分不必要的IO。
• 缺乏Spark集群配置优化：默认配置下Spark没法充分利用集群资源，尤其是处理S3这类对象存储时，需要针对性调优。
• 全量去重的开销：如果你的去重逻辑是基于特定列（比如示例里的Name、Class、Age、DOB），明确指定这些列去重比用distinct()更清晰，还能减少shuffle的数据量。
• 写回时没处理小文件：原分区有很多小文件，处理后如果直接写回去会更糟，需要合并小文件提升后续读取效率。

优化后的Python Spark代码 #

from pyspark.sql import SparkSession
import datetime

def main():
    # 你提供的Name和Age键值对列表
    target_name_age = [("Ram", 14), ("Shyam", 16)]
    
    # 初始化SparkSession，添加针对性配置
    dt_string = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    AppName = "ParquetDeduplication"
    
    spark = (SparkSession.builder
             .appName(f"{AppName}_{dt_string}")
             # 集群资源配置，根据你的集群规模调整
             .config("spark.executor.memory", "8g")
             .config("spark.executor.cores", "4")
             .config("spark.driver.memory", "4g")
             # S3优化配置
             .config("spark.hadoop.fs.s3a.multipart.size", "104857600")  # 100MB分块
             .config("spark.hadoop.fs.s3a.fast.upload", "true")
             # 调整shuffle并行度，避免过多小文件
             .config("spark.sql.shuffle.partitions", "200")
             .getOrCreate())
    
    spark.sparkContext.setLogLevel("ERROR")
    
    # 遍历每个目标分区，逐个处理（比读全路径更高效）
    for name, age in target_name_age:
        # 构造具体的S3分区路径
        s3_path = f"s3://bucket-name/{name}/{age}/date/"
        print(f"Processing partition: {s3_path}")
        
        # 读取分区数据
        df = spark.read.parquet(s3_path)
        
        # 基于指定列去重（替换成你的实际去重列）
        deduped_df = df.dropDuplicates(["Name", "Class", "Age", "DOB"])
        
        # 写回数据，保留原分区结构，合并小文件
        (deduped_df.write
         .mode("overwrite")
         .option("mergeSchema", "false")
         # 合并小文件，设置合理的单文件记录数（根据你的行大小调整）
         .option("maxRecordsPerFile", 100000)
         .parquet(s3_path))
        
        print(f"Finished deduplication for {s3_path}")
    
    spark.stop()

if __name__ == "__main__":
    main()

对应的Scala Spark代码 #

如果你的集群更适合用Scala，下面是等价的优化版本：

import org.apache.spark.sql.SparkSession
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

object ParquetDeduplication {
  def main(args: Array[String]): Unit = {
    // 目标Name和Age键值对
    val targetNameAge = List(("Ram", 14), ("Shyam", 16))
    
    val dtString = LocalDateTime.now.format(DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss"))
    val appName = s"ParquetDeduplication_$dtString"
    
    val spark = SparkSession.builder()
      .appName(appName)
      // 集群资源配置，按需调整
      .config("spark.executor.memory", "8g")
      .config("spark.executor.cores", "4")
      .config("spark.driver.memory", "4g")
      // S3优化配置
      .config("spark.hadoop.fs.s3a.multipart.size", "104857600")
      .config("spark.hadoop.fs.s3a.fast.upload", "true")
      // Shuffle并行度调整
      .config("spark.sql.shuffle.partitions", "200")
      .getOrCreate()
    
    spark.sparkContext.setLogLevel("ERROR")
    
    // 遍历每个分区处理
    targetNameAge.foreach { case (name, age) =>
      val s3Path = s"s3://bucket-name/$name/$age/date/"
      println(s"Processing partition: $s3Path")
      
      val df = spark.read.parquet(s3Path)
      
      // 指定去重列
      val dedupedDF = df.dropDuplicates(Seq("Name", "Class", "Age", "DOB"))
      
      // 写回数据，合并小文件
      dedupedDF.write
        .mode("overwrite")
        .option("mergeSchema", "false")
        .option("maxRecordsPerFile", 100000)
        .parquet(s3Path)
      
      println(s"Finished deduplication for $s3Path")
    }
    
    spark.stop()
  }
}

额外注意事项 #

1. 分区过滤的必要性：如果你的目标Name和Age组合很多，逐个处理分区比一次性读所有路径要高效得多，因为Spark可以精准定位到需要的文件，避免读取无关数据。
2. 资源配置调整：根据你的集群规模调整executor.memory和executor.cores，确保每个executor有足够的内存处理数据，避免OOM。
3. 去重列的选择：一定要明确指定参与去重的列，不要用默认的distinct()（虽然效果一样，但指定列更清晰，且如果有不需要参与去重的列，能减少shuffle的数据量）。
4. 小文件合并：通过maxRecordsPerFile或者spark.sql.files.maxPartitionBytes来控制输出文件的大小，避免生成大量小文件，影响后续读取性能。

备注：内容来源于stack exchange，提问作者kislay kashyap