WIP

Author: giwa

examples/src/main/python/streaming/test_oprations.py

@@ -15,10 +15,11 @@
 
     lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))
     words = lines.flatMap(lambda line: line.split(" "))
+#    ssc.checkpoint("checkpoint")
     mapped_words = words.map(lambda word: (word, 1))
     count = mapped_words.reduceByKey(add)
 
     count.pyprint()
     ssc.start()
-#    ssc.awaitTermination()
-    ssc.stop()
+    ssc.awaitTermination()
+#    ssc.stop()

python/pyspark/streaming/context.py

@@ -133,3 +133,8 @@ def stop(self, stopSparkContext=True):
         finally:
             # Stop Callback server
             SparkContext._gateway.shutdown()
+
+    def checkpoint(self, directory):
+        """
+        """
+        self._jssc.checkpoint(directory)

python/pyspark/streaming/dstream.py

@@ -172,7 +172,8 @@ def add_shuffle_key(split, iterator):
         with _JavaStackTrace(self.ctx) as st:
             partitioner = self.ctx._jvm.PythonPartitioner(numPartitions,
                                                       id(partitionFunc))
-            jdstream = self.ctx._jvm.PairwiseDStream(keyed._jdstream.dstream(), partitioner).asJavaDStream()
+            jdstream = self.ctx._jvm.PythonPairwiseDStream(keyed._jdstream.dstream(),
+                                                           partitioner).asJavaDStream()
         dstream = DStream(jdstream, self._ssc, BatchedSerializer(outputSerializer))
         # This is required so that id(partitionFunc) remains unique, even if
         # partitionFunc is a lambda:
@@ -233,6 +234,7 @@ def takeAndPrint(rdd, time):
     #    jdstream = self.ctx._jvm.PythonTransformedDStream(self._jdstream.dstream(), wrapped_func).toJavaDStream
     #    return DStream(jdstream, self._ssc, ...)  ## DO NOT KNOW HOW 
 
+
 class PipelinedDStream(DStream):
     def __init__(self, prev, func, preservesPartitioning=False):
         if not isinstance(prev, PipelinedDStream) or not prev._is_pipelinable():

python/pyspark/streaming/utils.py

@@ -37,7 +37,6 @@ class Java:
         implements = ['org.apache.spark.streaming.api.python.PythonRDDFunction']
 
 
-
 def msDurationToString(ms):
     """
     Returns a human-readable string representing a duration such as "35ms"

streaming/src/main/scala/org/apache/spark/streaming/api/python/PythonDStream.scala

@@ -25,7 +25,7 @@ import org.apache.spark._
 import org.apache.spark.rdd.RDD
 import org.apache.spark.api.python._
 import org.apache.spark.broadcast.Broadcast
-import org.apache.spark.streaming.{Duration, Time}
+import org.apache.spark.streaming.{StreamingContext, Duration, Time}
 import org.apache.spark.streaming.dstream._
 import org.apache.spark.streaming.api.java._
 
@@ -64,7 +64,7 @@ class PythonDStream[T: ClassTag](
 }
 
 
-private class PairwiseDStream(prev:DStream[Array[Byte]], partitioner: Partitioner) extends
+private class PythonPairwiseDStream(prev:DStream[Array[Byte]], partitioner: Partitioner) extends
 DStream[Array[Byte]](prev.ssc){
   override def dependencies = List(prev)
 
@@ -105,6 +105,7 @@ class PythonForeachDStream(
 
   this.register()
 }
+
 /*
 This does not work. Ignore this for now. -TD
 class PythonTransformedDStream(