fix accumulator with reused worker

Author: davies

python/pyspark/tests.py

@@ -1106,6 +1106,17 @@ def test_after_jvm_exception(self):
         rdd = self.sc.parallelize(range(100), 1)
         self.assertEqual(100, rdd.map(str).count())
 
+    def test_accumulator_when_reuse_worker(self):
+        from pyspark.accumulators import INT_ACCUMULATOR_PARAM
+        acc1 = self.sc.accumulator(0, INT_ACCUMULATOR_PARAM)
+        self.sc.parallelize(range(100), 20).foreach(lambda x: acc1.add(x))
+        self.assertEqual(sum(range(100)), acc1.value)
+
+        acc2 = self.sc.accumulator(0, INT_ACCUMULATOR_PARAM)
+        self.sc.parallelize(range(100), 20).foreach(lambda x: acc2.add(x))
+        self.assertEqual(sum(range(100)), acc2.value)
+        self.assertEqual(sum(range(100)), acc1.value)
+
 
 class TestSparkSubmit(unittest.TestCase):
 

python/pyspark/worker.py

@@ -76,6 +76,7 @@ def main(infile, outfile):
                 bid = - bid - 1
                 _broadcastRegistry.pop(bid, None)
 
+        _accumulatorRegistry.clear()
         command = pickleSer._read_with_length(infile)
         (func, deserializer, serializer) = command
         init_time = time.time()