如何在PySpark中高效清理Parquet文件特定行？（数据合规清除）

针对你在EMR PySpark环境下处理5年历史Parquet数据删除的性能问题，下面分享几个经过行业验证的高效方案和最佳实践，核心思路是避免全量重写，精准定位并只处理包含目标用户数据的资源：

1. 先通过Athena快速定位目标数据的位置 #

不要直接用Spark全量扫表，先借助Athena的Serverless查询能力，快速找出包含目标用户记录的Parquet文件路径和分区：

• 执行Athena查询获取目标资源：

  SELECT DISTINCT "$path", year, month 
  FROM your_athena_table 
  WHERE user_id = 'target_user_id'
  

  这个查询会利用Athena的元数据和分区扫描，几秒到几分钟就能返回结果，远快于Spark全量读取。
• 将查询结果导出到S3，后续用PySpark读取这个结果集，得到需要处理的文件/分区列表。

2. Spark增量处理：只操作目标文件/分区 #

拿到目标路径后，仅加载需要处理的Parquet文件，过滤后重写对应位置，避免全量计算：

• 示例PySpark代码：

  from pyspark.sql import SparkSession
  
  spark = SparkSession.builder.appName("UserRecordDeletion").getOrCreate()
  
  # 读取Athena查询结果，获取待处理文件路径
  target_files_df = spark.read.csv(
      "s3://your-athena-results-bucket/query-output/",
      header=True,
      inferSchema=True
  )
  target_files = target_files_df.select("$path").rdd.flatMap(lambda x: x).collect()
  
  # 开启Parquet优化，推动谓词下推
  spark.conf.set("spark.sql.parquet.filterPushdown", "true")
  spark.conf.set("spark.sql.parquet.enableVectorizedReader", "true")
  
  # 仅读取目标文件
  raw_df = spark.read.parquet(*target_files)
  # 过滤目标用户记录
  cleaned_df = raw_df.filter(raw_df.user_id != "target_user_id")
  
  # 按分区重写（避免覆盖整个分区，只更新有变化的文件）
  for file_path in target_files:
      # 解析分区目录（假设分区格式为s3://bucket/year=YYYY/month=MM/xxx.parquet）
      partition_parts = file_path.split("/")
      year_part = next(p for p in partition_parts if p.startswith("year=")).split("=")[1]
      month_part = next(p for p in partition_parts if p.startswith("month=")).split("=")[1]
      partition_dir = f"s3://your-bucket/year={year_part}/month={month_part}"
  
      # 过滤对应分区的数据并写入
      cleaned_df.filter(
          (cleaned_df.year == year_part) & (cleaned_df.month == month_part)
      ).write.mode("overwrite").parquet(partition_dir)
  

  这里的关键是只处理包含目标用户的文件，而非整个分区，大幅减少IO和计算开销。

3. 长期优化：从数据湖设计层面规避问题 #

如果这类删除请求频繁，建议从数据湖架构上提前优化：

• 按用户ID哈希分区：将user_id哈希取模后作为分区键（比如hash_user_id=0-99），删除时仅需处理对应哈希分区，无需遍历全量数据。
• CDC变更日志方案：用Debezium等工具记录数据变更，删除请求仅在变更日志中标记，查询时通过合并原始数据和变更日志获取最新状态，避免直接修改Parquet文件（适合读多写少的场景）。
• Lake Formation行级权限：若法规允许逻辑删除而非物理删除，可通过Lake Formation设置行级过滤，限制访问目标用户的记录，无需修改物理文件。
• 分层存储：将冷数据归档到S3 Glacier，删除时先通过Athena冰川查询定位数据，再按需解冻处理，降低存储成本和不必要的计算。

4. EMR集群配置优化 #

针对删除任务调整EMR集群参数，提升处理效率：

• 使用Spot实例降低成本，批量删除任务对可用性要求较低。
• 调整Spark资源配置，提高并行度：

  spark-submit --executor-memory 16G --executor-cores 4 --num-executors 20 your-deletion-script.py
  

• 开启S3 Transfer Acceleration，加快跨区域数据读写速度。

内容的提问来源于stack exchange，提问作者Carlos Eduardo Bilar Rodrigues