added TODO coments

YanTangZhai · Aug 16, 2014 · ea9c873 · ea9c873
1 parent 89ae38a
commit ea9c873
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 3 deletions.
diff --git a/python/pyspark/streaming/context.py b/python/pyspark/streaming/context.py
@@ -17,7 +17,6 @@
 
 import sys
 from signal import signal, SIGTERM, SIGINT
-from tempfile import NamedTemporaryFile
 
 from pyspark.serializers import PickleSerializer, BatchedSerializer, UTF8Deserializer
 from pyspark.context import SparkContext
@@ -79,6 +78,7 @@ def _clean_up_trigger(self):
         """Kill py4j callback server properly using signal lib"""
 
         def clean_up_handler(*args):
+            SparkContext._gateway._shutdown_callback_server()
             SparkContext._gateway.shutdown()
             sys.exit(0)
 
@@ -128,6 +128,7 @@ def stop(self, stopSparkContext=True, stopGraceFully=False):
             self._jssc.stop(stopSparkContext, stopGraceFully)
         finally:
             # Stop Callback server
+            SparkContext._gateway._shutdown_callback_server()
             SparkContext._gateway.shutdown()
 
     def _testInputStream(self, test_inputs, numSlices=None):

diff --git a/python/pyspark/streaming/dstream.py b/python/pyspark/streaming/dstream.py
@@ -376,15 +376,27 @@ def saveAsTextFile(rdd, time):
         return self.foreachRDD(saveAsTextFile)
 
 
+# TODO: implement updateStateByKey
+# TODO: implement slice
+
+# Window Operations
+# TODO: implement window
+# TODO: implement groupByKeyAndWindow
+# TODO: implement reduceByKeyAndWindow
+# TODO: implement countByValueAndWindow
+# TODO: implement countByWindow
+# TODO: implement reduceByWindow
+
 # Following operation has dependency to transform
-# TODO: impelment union
+# TODO: implement transform
+# TODO: implement transformWith
+# TODO: implement union
 # TODO: implement repertitions
 # TODO: implement cogroup
 # TODO: implement join
 # TODO: implement leftOuterJoin
 # TODO: implemtnt rightOuterJoin
 
-
 class PipelinedDStream(DStream):
     def __init__(self, prev, func, preservesPartitioning=False):
         if not isinstance(prev, PipelinedDStream) or not prev._is_pipelinable():