Spark 2.4.7作业中，复用缓存DataFrame后如何取消持久化？

在你的场景中，直接在C函数末尾调用cachedDf.unpersist()是无效的——因为Spark的转换操作是懒执行的，此时依赖cachedDf的join和unionByName还没有真正执行，提前释放缓存会导致后续计算重新生成原始数据，完全失去缓存的意义。

你需要确保缓存的释放时机是在所有依赖cachedDf的操作执行完成之后，以下是两种可靠的实现方式：

方法1：利用Dataset.onCompletion注册回调（推荐） #

Spark 2.3+提供了onCompletion方法，可以在Dataset对应的行动操作完成后触发回调逻辑。你可以在生成最终结果DataFrame后，注册一个回调来释放缓存：

def C(): DataFrame => DataFrame = inputDf => {
  val cachedDf = inputDf.cache()

  val df1 = cachedDf.transform(...)
  val df2 = ... // 你的df2生成逻辑

  val resultDf = cachedDf.join(df2).unionByName(df1)

  // 注册回调：当resultDf的行动操作（比如后续的write）完成后，释放缓存
  resultDf.onCompletion { _ =>
    cachedDf.unpersist()
    // 如果需要强制立即释放（跳过等待GC），可以用：cachedDf.unpersist(true)
  }

  resultDf
}

这种方式的优势是不需要额外传递缓存引用，回调会自动在依赖操作完成后执行，完全适配链式transform的代码结构。

方法2：在作业最终行动操作后手动释放 #

如果你需要更明确的控制，可以将cachedDf的引用传递到作业末尾，在write完成后执行unpersist。但这种方式需要调整代码结构，比如将cachedDf作为外部变量或者返回值传递：

// 调整C函数，返回结果DataFrame和缓存的DataFrame
def C(): DataFrame => (DataFrame, DataFrame) = inputDf => {
  val cachedDf = inputDf.cache()

  val df1 = cachedDf.transform(...)
  val df2 = ...

  val resultDf = cachedDf.join(df2).unionByName(df1)
  (resultDf, cachedDf)
}

// 调用时：
val (afterC, cachedDf) = InputDf.transform(A).transform(B).transform(C())
afterC.transform(D)...transform(Z)
  .write.format("xxx")
  .mode(saveMode)
  .save(Path)

// 行动操作完成后释放缓存
cachedDf.unpersist()

这种方式适合需要精确控制释放时机的场景，但会打破链式调用的简洁性。

注意事项 #

• 不要在转换阶段（未触发行动操作前）调用unpersist，否则缓存会提前失效。
• 如果你的df2生成逻辑也依赖cachedDf，上述两种方法都能确保所有依赖操作完成后再释放缓存。
• 使用unpersist(true)会强制立即释放内存中的缓存，而不带参数的unpersist()会标记缓存为待释放，由GC自动清理，根据你的资源情况选择即可。

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