Switch cast in PairDStreamFunctions

pdeyhim · Apr 8, 2014 · 4b0eeb9 · 4b0eeb9
1 parent fa395c9
commit 4b0eeb9
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/dstream/PairDStreamFunctions.scala b/streaming/src/main/scala/org/apache/spark/streaming/dstream/PairDStreamFunctions.scala
@@ -72,7 +72,7 @@ class PairDStreamFunctions[K: ClassTag, V: ClassTag](self: DStream[(K,V)])
     val mergeValue = (c: ArrayBuffer[V], v: V) => (c += v)
     val mergeCombiner = (c1: ArrayBuffer[V], c2: ArrayBuffer[V]) => (c1 ++ c2)
     combineByKey(createCombiner, mergeValue, mergeCombiner, partitioner)
-      .asInstanceOf[DStream[(K, Iterator[V])]]
+      .asInstanceOf[DStream[(K, Seq[V])]].mapValues(_.toIterator)
   }
 
   /**
@@ -188,7 +188,7 @@ class PairDStreamFunctions[K: ClassTag, V: ClassTag](self: DStream[(K,V)])
     self.groupByKey(partitioner)
         .window(windowDuration, slideDuration)
         .combineByKey[ArrayBuffer[V]](createCombiner, mergeValue, mergeCombiner, partitioner)
-        .asInstanceOf[DStream[(K, Iterator[V])]]
+        .asInstanceOf[DStream[(K, Seq[V])]].mapValues(_.toIterator)
   }
 
   /**