如何确保垃圾收集未使用的蓄电池？

Question

我遇到一个问题，即Spark上的Accumulator不能是GC。

def newIteration (lastParams: Accumulable[Params, (Int, Int, Int)], lastChosens: RDD[Document], i: Int): Params = { 
    if (i == maxIteration) 
     return lastParams.value 

    val size1: Int = 100 
    val size2: Int = 1000 

    // each iteration generates a new accumulator 
    val params = sc.accumulable(Params(size1, size2)) 

    // there is map operation here 
    // if i only use lastParams, the result in not updated 
    // but params can solve this problem 
    val chosen = data.map { 
     case(Document(docID, content)) => { 
      lastParams += (docID, content, -1) 
      val newContent = lastParams.localValue.update(docID, content) 
      lastParams += (docID, newContent, 1) 
      params += (docID, newContent, 1) 
      Document(docID, newContent) 
     } 
    }.cache() 
    chosen.count() 
    lastChosens.unpersist() 
    return newIteration(params, chosen, i + 1) 
} 

问题是，它分配的内存一直在增长，直到内存限制。看来lastParms不是GC。类RDD和Broadcast有一个方法unpersist()，但我在文档中找不到像这样的任何方法。

为什么Accumulable不能自动GC，还是有更好的解决方案？

Answer 1

更新（2016年4月22日）：SPARK-3885 Provide mechanism to remove accumulators once they are no longer used现已解决。

目前正在进行的工作是在不再引用自动垃圾收集累加器时添加支持。请参阅SPARK-3885以跟踪此功能的进度。目前正在审查中的Spark PR #4021是此功能的补丁。我期望将其包含在Spark 1.3.0中。