4. 缓存清除：unpersist()的时机与最佳实践，避免内存泄漏

缓存用完了，不清理会怎样？

说白了，就是内存泄漏。Spark 的缓存数据占着 Executor 内存不释放，后续的 Shuffle、Join 操作反而没内存可用，性能直接崩掉。我见过太多同学只记得 .cache()，从来不调 .unpersist()，结果集群跑着跑着就 OOM 了。

嗯，这一章我们就聊聊：什么时候该清缓存？怎么清才干净？

4.1 为什么必须手动清除缓存？

Spark 的 cache() 和 persist() 只是把数据标记为“可缓存”，真正写入存储层是在 Action 触发时。同样，unpersist() 也只是标记“可删除”，但删除动作是异步的。

这里有个关键点：Spark 不会自动清理不再使用的缓存。除非你显式调用 unpersist()，或者缓存数据因为内存压力被 LRU 淘汰。但 LRU 淘汰不可控，你无法保证它先淘汰谁。

4.2 unpersist() 的核心机制

先看 API 签名：

def unpersist(blocking: Boolean = false): Dataset.this.type

blocking 参数控制是否阻塞等待数据真正删除：

我个人习惯用默认的异步模式。除非你马上要执行一个内存密集型的 Shuffle，否则没必要等它删完。

4.3 最佳实践：什么时候该 unpersist？

我总结了 4 个黄金时机：

1. 数据不再被后续计算引用时 —— 最直接的规则。比如你 cache 了一个中间表，后面所有计算都基于它的结果，那就可以在最终 Action 之后 unpersist。
2. 迭代计算中，每轮结束后 —— 比如机器学习迭代、图计算，每轮生成新的缓存，旧缓存必须清掉。
3. 缓存数据被重新计算覆盖前 —— 如果你要对同一个 DataFrame 换一种存储级别重新 persist，先 unpersist 旧的。
4. 作业结束前 —— 即使 SparkContext 停止时会自动清理，但显式 unpersist 能让内存更早释放，方便同一集群上的其他作业。

4.4 代码示例：正确的 unpersist 模式

// 错误示范：只 cache 不 unpersist
val df = spark.read.parquet("hdfs://data/sales")
df.cache()
df.count()
val result = df.groupBy("city").agg(sum("amount"))
result.write.mode("overwrite").parquet("hdfs://output/sales_summary")
// df 的缓存还在内存里，白白占着 2GB

// 正确示范：成对出现
val df = spark.read.parquet("hdfs://data/sales")
df.cache()
try {
  df.count()
  val result = df.groupBy("city").agg(sum("amount"))
  result.write.mode("overwrite").parquet("hdfs://output/sales_summary")
} finally {
  df.unpersist()  // 确保无论是否异常，缓存都被清除
}

用 try-finally 包裹是个好习惯。万一中间代码抛异常，缓存也能被释放。

4.5 批量清除与选择性清除

有时候你 cache 了多个 DataFrame，想一次性清掉：

// 批量清除
Seq(df1, df2, df3).foreach(_.unpersist())

// 或者用 SparkContext 清除所有缓存
spark.catalog.clearCache()

clearCache() 会清除当前 SparkSession 中所有缓存的表和 DataFrame。我建议只在开发调试时用，生产环境还是精确控制比较好。

4.6 避坑指南：unpersist 的常见误区

• 误区一：unpersist 后 DataFrame 还能用 —— 不行。unpersist 只是删除缓存数据，DataFrame 的元数据还在，但再次触发 Action 时会重新计算。如果你不想重新计算，就别 unpersist。
• 误区二：unpersist 立即释放内存 —— 异步模式下不会。JVM 的 GC 有自己的节奏，unpersist 只是把数据从 Storage Memory 标记为可回收，真正释放要看 GC 时机。
• 误区三：多次 unpersist 没问题 —— 没问题，但多余。第二次调用只是空操作，不会报错，但浪费一次 RPC 开销。
• 误区四：用 df.createOrReplaceTempView() 后 unpersist 会删表 —— 不会。unpersist 只影响缓存，临时视图依然存在，只是下次查询时不再从缓存读。

4.7 知识体系：缓存清除决策流程

下面这张图总结了我平时做缓存清除的决策逻辑：

4.8 监控缓存状态

你可以在 Spark UI 的 Storage 标签页看到当前所有缓存的数据。我建议养成习惯：作业跑完后看一眼 Storage 页，确认缓存都清干净了。

也可以用代码检查：

// 查看当前缓存的数据集
spark.catalog.cachedDataFrames.foreach { df =>
  println(s"缓存表: ${df.queryExecution.analyzed.treeString.take(50)}")
}

// 查看缓存占用的内存大小（近似）
val storageStatus = spark.sharedState.cacheManager.cacheStats
println(s"当前缓存大小: ${storageStatus.memSize} bytes")

好了，这一章的内容就这些。记住：cache() 和 unpersist() 是成对出现的，就像锁和释放锁一样。养成这个习惯，你的 Spark 作业就能远离内存泄漏的烦恼。