make accumulator.localValue public, add tests

Imran Rashid · Imran Rashid · commit 206a3833ce17 · 2012-08-14T14:08:22.000-07:00
diff --git a/core/src/main/scala/spark/Accumulators.scala b/core/src/main/scala/spark/Accumulators.scala
@@ -35,7 +35,16 @@ class Accumulable[T,R] (
     else throw new UnsupportedOperationException("Can't use read value in task")
   }
 
-  private[spark] def localValue = value_
+  /**
+   * get the current value of this accumulator from within a task.
+   *
+   * This is NOT the global value of the accumulator.  To get the global value after a
+   * completed operation on the dataset, call `value`.
+   *
+   * The typical use of this method is to directly mutate the local value, eg., to add
+   * an element to a Set.
+   */
+  def localValue = value_
 
   def value_= (t: T) {
     if (!deserialized) value_ = t
diff --git a/core/src/test/scala/spark/AccumulatorSuite.scala b/core/src/test/scala/spark/AccumulatorSuite.scala
@@ -79,4 +79,18 @@ class AccumulatorSuite extends FunSuite with ShouldMatchers {
     }
   }
 
+  test ("localValue readable in tasks") {
+    import SetAccum._
+    val maxI = 1000
+    for (nThreads <- List(1, 10)) { //test single & multi-threaded
+    val sc = new SparkContext("local[" + nThreads + "]", "test")
+      val acc: Accumulable[mutable.Set[Any], Any] = sc.accumulable(new mutable.HashSet[Any]())
+      val d = sc.parallelize(1 to maxI)
+      d.foreach {
+        x => acc.localValue += x
+      }
+      acc.value should be ( (1 to maxI).toSet)
+    }
+  }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -79,4 +79,18 @@ class AccumulatorSuite extends FunSuite with ShouldMatchers {`
`79`	`79`	`}`
`80`	`80`	`}`
`81`	`81`
	`82`	`+ test ("localValue readable in tasks") {`
	`83`	`+ import SetAccum._`
	`84`	`+ val maxI = 1000`
	`85`	`+ for (nThreads <- List(1, 10)) { //test single & multi-threaded`
	`86`	`+ val sc = new SparkContext("local[" + nThreads + "]", "test")`
	`87`	`+ val acc: Accumulable[mutable.Set[Any], Any] = sc.accumulable(new mutable.HashSet[Any]())`
	`88`	`+ val d = sc.parallelize(1 to maxI)`
	`89`	`+ d.foreach {`
	`90`	`+ x => acc.localValue += x`
	`91`	`+ }`
	`92`	`+ acc.value should be ( (1 to maxI).toSet)`
	`93`	`+ }`
	`94`	`+ }`
	`95`	`+`
`82`	`96`	`}`