Updates from matei's review

Author: srowen

core/src/main/scala/org/apache/spark/partial/GroupedCountEvaluator.scala

@@ -39,15 +39,15 @@ private[spark] class GroupedCountEvaluator[T : ClassTag](totalOutputs: Int, conf
 
   override def merge(outputId: Int, taskResult: OpenHashMap[T,Long]) {
     outputsMerged += 1
-    taskResult.foreach{ case (key,value) =>
+    taskResult.foreach { case (key, value) =>
       sums.changeValue(key, value, _ + value)
     }
   }
 
   override def currentResult(): Map[T, BoundedDouble] = {
     if (outputsMerged == totalOutputs) {
       val result = new JHashMap[T, BoundedDouble](sums.size)
-      sums.foreach{ case (key,sum) =>
+      sums.foreach { case (key, sum) =>
         result(key) = new BoundedDouble(sum, 1.0, sum, sum)
       }
       result
@@ -57,7 +57,7 @@ private[spark] class GroupedCountEvaluator[T : ClassTag](totalOutputs: Int, conf
       val p = outputsMerged.toDouble / totalOutputs
       val confFactor = Probability.normalInverse(1 - (1 - confidence) / 2)
       val result = new JHashMap[T, BoundedDouble](sums.size)
-      sums.foreach{ case (key, sum) =>
+      sums.foreach { case (key, sum) =>
         val mean = (sum + 1 - p) / p
         val variance = (sum + 1) * (1 - p) / (p * p)
         val stdev = math.sqrt(variance)

core/src/main/scala/org/apache/spark/rdd/RDD.scala

@@ -836,13 +836,13 @@ abstract class RDD[T: ClassTag](
     // TODO: This should perhaps be distributed by default.
     def countPartition(iter: Iterator[T]): Iterator[OpenHashMap[T,Long]] = {
       val map = new OpenHashMap[T,Long]
-      iter.foreach{
+      iter.foreach {
         t => map.changeValue(t, 1L, _ + 1L)
       }
       Iterator(map)
     }
     def mergeMaps(m1: OpenHashMap[T,Long], m2: OpenHashMap[T,Long]): OpenHashMap[T,Long] = {
-      m2.foreach{ case (key, value) =>
+      m2.foreach { case (key, value) =>
         m1.changeValue(key, value, _ + value)
       }
       m1
@@ -865,7 +865,7 @@ abstract class RDD[T: ClassTag](
     }
     val countPartition: (TaskContext, Iterator[T]) => OpenHashMap[T,Long] = { (ctx, iter) =>
       val map = new OpenHashMap[T,Long]
-      iter.foreach{
+      iter.foreach {
         t => map.changeValue(t, 1L, _ + 1L)
       }
       map

core/src/main/scala/org/apache/spark/util/collection/AppendOnlyMap.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.util.collection
 
 import java.util.{Arrays, Comparator}
+
 import com.google.common.hash.Hashing
 
 import org.apache.spark.annotation.DeveloperApi

core/src/main/scala/org/apache/spark/util/collection/OpenHashMap.scala

@@ -30,6 +30,7 @@ import org.apache.spark.annotation.DeveloperApi
  * Under the hood, it uses our OpenHashSet implementation.
  */
 @DeveloperApi
+private[spark]
 class OpenHashMap[K : ClassTag, @specialized(Long, Int, Double) V: ClassTag](
     initialCapacity: Int)
   extends Iterable[(K, V)]

streaming/src/main/scala/org/apache/spark/streaming/util/RawTextHelper.scala

@@ -30,11 +30,30 @@ object RawTextHelper {
    */
   def splitAndCountPartitions(iter: Iterator[String]): Iterator[(String, Long)] = {
     val map = new OpenHashMap[String,Long]
-    val tokenized = iter.flatMap(_.split(" ").filterNot(_.isEmpty))
-    tokenized.foreach{ s =>
-      map.changeValue(s, 1L, _ + 1L)
+    var i = 0
+    var j = 0
+    while (iter.hasNext) {
+      val s = iter.next()
+      i = 0
+      while (i < s.length) {
+        j = i
+        while (j < s.length && s.charAt(j) != ' ') {
+          j += 1
+        }
+        if (j > i) {
+          val w = s.substring(i, j)
+          map.changeValue(w, 1L, _ + 1L)
+        }
+        i = j
+        while (i < s.length && s.charAt(i) == ' ') {
+          i += 1
+        }
+      }
+      map.toIterator.map {
+        case (k, v) => (k, v)
+      }
     }
-    map.iterator
+    map.toIterator.map{case (k, v) => (k, v)}
   }
 
   /**