Clear local copies of accumulators as soon as we're done with them

Author: Nathan Kronenfeld

core/src/main/scala/org/apache/spark/Accumulators.scala

@@ -311,6 +311,7 @@ private object Accumulators {
     for ((id, accum) <- localAccums.getOrElse(Thread.currentThread, Map())) {
       ret(id) = accum.localValue
     }
+    localAccums.remove(Thread.currentThread)
     return ret
   }