Spark执行计划多persist引发磁盘不足的优化方案问询

你的推测完全成立 #

Spark的persist操作会将当前DataFrame的计算结果缓存到内存或磁盘中，默认是惰性求值——直到触发count()、show()这类action操作时，所有标记过的persist才会实际执行并存储数据。

由于你在代码中多次调用persist且未主动清理，这些缓存的数据会持续占用磁盘空间（当内存不足时）。当所有缓存的总大小超过集群可用磁盘空间时，就会触发No space left on device错误。从explain()输出能看到多个persist节点，也验证了这些缓存操作都会被执行。

如何在第N次Persist时释放第N-1次的缓存 #

核心思路是：在不需要前一阶段的缓存DataFrame时，主动调用unpersist()释放其占用的资源。注意不要复用同一个变量名覆盖前一阶段的DataFrame，否则会丢失引用无法清理。

推荐代码结构 #

# 第一阶段转换与缓存
stage1_df = spark.read.csv(s3path)
stage1_df = stage1_df.transformation1()
stage1_df = stage1_df.transformation2()
stage1_df = stage1_df.transformation3()
stage1_df = stage1_df.transformation4()
stage1_df.persist(MEMORY_AND_DISK)

# 基于stage1生成stage2，完成后释放stage1缓存
stage2_df = stage1_df.transformation5()
stage2_df = stage2_df.transformation6()
stage2_df = stage2_df.transformation7()
stage1_df.unpersist(blocking=True)  # blocking=True确保缓存清理完成再继续
stage2_df.persist(MEMORY_AND_DISK)

# 后续阶段以此类推
stage3_df = stage2_df.transformation8()
stage3_df = stage3_df.transformation9()
stage3_df = stage3_df.transformation10()
stage2_df.unpersist(blocking=True)
stage3_df.persist(MEMORY_AND_DISK)

如果一定要复用变量，需要先保存前一阶段的引用：

df = spark.read.csv(s3path)
df = df.transformation1()
df = df.transformation2()
df = df.transformation3()
df = df.transformation4()
df.persist(MEMORY_AND_DISK)

# 保存当前缓存的DataFrame引用
prev_stage_df = df
# 执行下一阶段转换
df = prev_stage_df.transformation5()
df = df.transformation6()
df = df.transformation7()
# 释放前一阶段缓存
prev_stage_df.unpersist(blocking=True)
df.persist(MEMORY_AND_DISK)

代码结构最佳实践 #

1. 按需Persist，不滥用：只有当某个DataFrame会被多次重复使用（比如作为多个后续转换的输入）时才需要缓存。如果是线性的单路径转换（每个阶段只被用一次），完全不需要persist——Spark的执行计划会自动优化，不会重复计算。
2. 选择合适的缓存级别：
   • 优先用MEMORY_ONLY：如果数据能完全放入内存，性能最优。
   • 内存不足时用MEMORY_ONLY_SER或MEMORY_AND_DISK_SER：序列化后的对象占用空间远小于非序列化，能大幅降低磁盘占用。
   • 避免盲目使用MEMORY_AND_DISK：非序列化的磁盘存储占用空间大，容易触发磁盘不足问题。
3. 主动清理缓存：在确认某个阶段的DataFrame不再被使用后，立即调用unpersist()释放资源，尤其是在磁盘资源有限的集群环境中。
4. 使用清晰的变量命名：用stage1_df、stage2_df这类命名区分不同阶段的DataFrame，避免引用丢失，同时提升代码可读性。
5. 监控缓存状态：通过Spark UI的Storage页面查看缓存的Data列表、占用空间，快速定位缓存溢出问题。

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