Flume example and bug fix

1 files changed, 7 insertions, 2 deletions

diff --git a/streaming/src/main/scala/spark/streaming/api/java/JavaDStreamLike.scala b/streaming/src/main/scala/spark/streaming/api/java/JavaDStreamLike.scala
index cb58c1351d..91bcca9afa 100644
--- a/streaming/src/main/scala/spark/streaming/api/java/JavaDStreamLike.scala
+++ b/streaming/src/main/scala/spark/streaming/api/java/JavaDStreamLike.scala
@@ -1,6 +1,7 @@
 package spark.streaming.api.java
 
 import java.util.{List => JList}
+import java.lang.{Integer => JInt}
 
 import scala.collection.JavaConversions._
 
@@ -16,6 +17,10 @@ trait JavaDStreamLike[T, This <: JavaDStreamLike[T, This]] extends Serializable
 
   def dstream: DStream[T]
 
+  implicit def scalaIntToJavaInteger(in: DStream[Int]): JavaDStream[JInt] = {
+    in.map(new JInt(_))
+  }
+
   /**
    * Prints the first ten elements of each RDD generated in this DStream. This is an output
    * operator, so this DStream will be registered as an output stream and there materialized.
@@ -26,14 +31,14 @@ trait JavaDStreamLike[T, This <: JavaDStreamLike[T, This]] extends Serializable
    * Returns a new DStream in which each RDD has a single element generated by counting each RDD
    * of this DStream.
    */
-  def count(): JavaDStream[Int] = dstream.count()
+  def count(): JavaDStream[JInt] = dstream.count()
 
   /**
    * Returns a new DStream in which each RDD has a single element generated by counting the number
    * of elements in a window over this DStream. windowDuration and slideDuration are as defined in the
    * window() operation. This is equivalent to window(windowDuration, slideDuration).count()
    */
-  def countByWindow(windowDuration: Duration, slideDuration: Duration) : JavaDStream[Int] = {
+  def countByWindow(windowDuration: Duration, slideDuration: Duration) : JavaDStream[JInt] = {
     dstream.countByWindow(windowDuration, slideDuration)
   }