如何确保未使用的累加器的垃圾收集?
How to ensure garbage collection of unused accumulators?
我遇到了一个问题,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。 Class RDD
和 Broadcast
有一个方法 unpersist()
,但我在文档中找不到任何类似的方法。
为什么Accumulable
不能自动GC,或者有更好的解决方案吗?
更新(2016 年 4 月 22 日):SPARK-3885 Provide mechanism to remove accumulators once they are no longer used 现已解决。
正在进行的工作是添加对不再被引用的自动垃圾收集累加器的支持。请参阅 SPARK-3885 for tracking progress on this feature. Spark PR #4021,目前正在审核中,是此功能的补丁。我希望它包含在 Spark 1.3.0 中。
我遇到了一个问题,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。 Class RDD
和 Broadcast
有一个方法 unpersist()
,但我在文档中找不到任何类似的方法。
为什么Accumulable
不能自动GC,或者有更好的解决方案吗?
更新(2016 年 4 月 22 日):SPARK-3885 Provide mechanism to remove accumulators once they are no longer used 现已解决。
正在进行的工作是添加对不再被引用的自动垃圾收集累加器的支持。请参阅 SPARK-3885 for tracking progress on this feature. Spark PR #4021,目前正在审核中,是此功能的补丁。我希望它包含在 Spark 1.3.0 中。