大规模PySpark DataFrame超资源限制下写入Parquet的代码优化问询

我有一个包含2.5亿行、仅2列的大规模PySpark DataFrame，正在运行MinHash相关代码。尝试执行以下代码将结果DataFrame写入Parquet文件时，持续遇到错误：

pod ephemeral local storage usage exceeds the total limit of containers

由于无法增加集群资源，希望通过优化PySpark代码解决该问题。目前已执行以下操作：

1. 运行MinHash函数前对DataFrame进行分区：sdf = sdf.repartition(200)
2. 在涉及两次Join的最终步骤前过滤掉不太可能共享大量哈希值的对：filtered_sdf = hash_sdf.filter(f.size(f.col('nodeSet')) > threshold)，其中threshold = int(0.2 * n_draws)
3. 设置Shuffle分区数：spark.conf.set("spark.sql.shuffle.partitions", "200")

请问还有哪些方法可以实现无资源问题的Parquet文件写入？

优化方案 #

1. 细化写入阶段的分区策略 #

当前200个分区可能导致单分区数据量过大，尤其是MinHash处理后可能存在隐性数据倾斜。可以：

• 基于数据分布更均匀的字段（比如某一列的哈希值）增加分区数，建议设置为集群核心数的2-4倍：

  # 假设其中一列名为"entity_id"，基于该列哈希拆分至800个分区
  adj_sdf = adj_sdf.repartition(800, f.hash("entity_id"))
  

• 限制单个Parquet文件的记录数，避免单文件占用过多临时存储：

  adj_sdf.write.mode("append") \
    .option("maxRecordsPerFile", 1000000)  # 每个文件最多100万条记录
    .parquet("/output/folder/")
  

2. 优化Shuffle阶段的磁盘占用 #

Pod存储超限大多源于Shuffle临时文件，调整以下配置减少磁盘压力：

# 开启Shuffle溢出数据压缩，降低临时文件体积
spark.conf.set("spark.shuffle.spill.compress", "true")
spark.conf.set("spark.io.compression.codec", "snappy")  # 用Snappy平衡压缩比和速度

# 调整Shuffle内存占比，减少不必要的磁盘溢出
spark.conf.set("spark.shuffle.memoryFraction", "0.4")
# 设置强制溢出阈值，避免小数据频繁写入磁盘
spark.conf.set("spark.shuffle.spill.numElementsForceSpillThreshold", "1000000")

3. 主动释放中间数据资源 #

在写入最终结果前，清理不再使用的中间DataFrame，释放内存和临时存储：

# 解除中间DataFrame的持久化
hash_sdf.unpersist()
filtered_sdf.unpersist()
# 触发JVM垃圾回收，释放闲置资源
spark.sparkContext._jvm.System.gc()

4. 优化Parquet写入的存储参数 #

通过Parquet的配置选项减少写入过程中的临时文件大小：

adj_sdf.write.mode("append") \
  .option("compression", "snappy")  # 启用文件压缩
  .option("parquet.block.size", "134217728")  # 设置块大小为128MB，匹配HDFS块大小
  .option("parquet.page.size", "1048576")  # 设置页大小为1MB，优化读写效率
  .parquet("/output/folder/")

5. 排查并解决数据倾斜 #

即使做了前置过滤，仍可能存在个别分区数据量远超均值的情况：

• 先检查分区数据分布：

  adj_sdf.groupBy(f.spark_partition_id()).count().orderBy("count", ascending=False).show(10)
  

• 若发现倾斜，对倾斜键添加随机后缀拆分分区：

  # 假设倾斜列是"pair_key"，添加0-9的随机后缀拆分
  adj_sdf = adj_sdf.withColumn("rand_tag", f.floor(f.rand() * 10)) \
    .repartition(800, "pair_key", "rand_tag") \
    .drop("rand_tag")
  

6. 调整Executor临时存储配置 #

如果集群允许调整Pod存储参数（不增加总资源，仅调整分配比例）：

# 设置每个Executor的临时存储上限，匹配Pod的存储配额
spark.conf.set("spark.executor.ephemeralStorage.size", "20g")
# 指定临时存储目录到挂载的额外磁盘（若有）
spark.conf.set("spark.local.dir", "/mnt/ephemeral-storage")

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