Merge pull request apache#173 from squito/accum_localValue

mateiz · mateiz · commit 79c82b6cfd11 · 2012-08-22T00:11:21.000-07:00
make accumulator.localValue public, add tests
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,19 @@ 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 groupedInts = (1 to (maxI/20)).map {x => (20 * (x - 1) to 20 * x).toSet}
+      val d = sc.parallelize(groupedInts)
+      d.foreach {
+        x => acc.localValue ++= x
+      }
+      acc.value should be ( (0 to maxI).toSet)
+    }
+  }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -79,4 +79,19 @@ 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 groupedInts = (1 to (maxI/20)).map {x => (20 * (x - 1) to 20 * x).toSet}`
	`89`	`+ val d = sc.parallelize(groupedInts)`
	`90`	`+ d.foreach {`
	`91`	`+ x => acc.localValue ++= x`
	`92`	`+ }`
	`93`	`+ acc.value should be ( (0 to maxI).toSet)`
	`94`	`+ }`
	`95`	`+ }`
	`96`	`+`
`82`	`97`	`}`