garbage-collection - 如何确保未使用的累加器的垃圾收集？

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 中。