title: How to ensure garbage collection of unused accumulators?
body: <p>I meet a problem that Accumulator on Spark can not be GC.</p>
  
  <pre><code>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)) =&gt; {
              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)
  }
  </code></pre>
  
  <p>The problem is that the memory it allocates is always growing, until memory limits. It seems that <code>lastParms</code> is not GC. Class <code>RDD</code> and <code>Broadcast</code> have a method <code>unpersist()</code>, but I cannot find any method like this in documentation.</p>
  
  <p>Why <code>Accumulable</code> cannot be GC automatically, or is there a better solution?</p>