change tests to show utility of localValue

Imran Rashid · Imran Rashid · commit 4d2efe9555f5 · 2012-08-20T15:17:31.000-07:00
diff --git a/core/src/main/scala/spark/Accumulators.scala b/core/src/main/scala/spark/Accumulators.scala
@@ -36,7 +36,7 @@ class Accumulable[T,R] (
   }
 
   /**
-   * get the current value of this accumulator from within a task.
+   * 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`.
diff --git a/core/src/test/scala/spark/AccumulatorSuite.scala b/core/src/test/scala/spark/AccumulatorSuite.scala
@@ -85,11 +85,12 @@ class AccumulatorSuite extends FunSuite with ShouldMatchers {
     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)
+      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
+        x => acc.localValue ++= x
       }
-      acc.value should be ( (1 to maxI).toSet)
+      acc.value should be ( (0 to maxI).toSet)
     }
   }
 

Original file line number	Diff line number	Diff line change
`@@ -36,7 +36,7 @@ class Accumulable[T,R] (`
`36`	`36`	`}`
`37`	`37`
`38`	`38`	`/**`
`39`		`- * get the current value of this accumulator from within a task.`
	`39`	`+ * Get the current value of this accumulator from within a task.`
`40`	`40`	`*`
`41`	`41`	`* This is NOT the global value of the accumulator. To get the global value after a`
`42`	`42`	* completed operation on the dataset, call `value`.
Original file line number	Diff line number	Diff line change
`@@ -85,11 +85,12 @@ class AccumulatorSuite extends FunSuite with ShouldMatchers {`
`85`	`85`	`for (nThreads <- List(1, 10)) { //test single & multi-threaded`
`86`	`86`	`val sc = new SparkContext("local[" + nThreads + "]", "test")`
`87`	`87`	`val acc: Accumulable[mutable.Set[Any], Any] = sc.accumulable(new mutable.HashSet[Any]())`
`88`		`- val d = sc.parallelize(1 to maxI)`
	`88`	`+ val groupedInts = (1 to (maxI/20)).map {x => (20 * (x - 1) to 20 * x).toSet}`
	`89`	`+ val d = sc.parallelize(groupedInts)`
`89`	`90`	`d.foreach {`
`90`		`- x => acc.localValue += x`
	`91`	`+ x => acc.localValue ++= x`
`91`	`92`	`}`
`92`		`- acc.value should be ( (1 to maxI).toSet)`
	`93`	`+ acc.value should be ( (0 to maxI).toSet)`
`93`	`94`	`}`
`94`	`95`	`}`
`95`	`96`