fixed for socketTextStream

Author: Ken Takagiwa

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

@@ -10,7 +10,7 @@
         exit(-1)
     ssc = StreamingContext(appName="PythonStreamingNetworkWordCount", duration=Seconds(1))
 
-    lines = ssc.socketTextStream(sys.argv[1], sys.argv[2])
+    lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))
     fm_lines = lines.flatMap(lambda x: x.split(" "))
     filtered_lines = fm_lines.filter(lambda line: "Spark" in line)
     mapped_lines = fm_lines.map(lambda x: (x, 1))

python/pyspark/java_gateway.py

@@ -85,6 +85,7 @@ def run(self):
     java_import(gateway.jvm, "org.apache.spark.streaming.*")
     java_import(gateway.jvm, "org.apache.spark.streaming.api.java.*")
     java_import(gateway.jvm, "org.apache.spark.streaming.api.python.*")
+    java_import(gateway.jvm, "org.apache.spark.streaming.dstream.*")
     java_import(gateway.jvm, "org.apache.spark.mllib.api.python.*")
     java_import(gateway.jvm, "org.apache.spark.sql.SQLContext")
     java_import(gateway.jvm, "org.apache.spark.sql.hive.HiveContext")

python/pyspark/streaming/context.py

@@ -19,7 +19,7 @@
 from pyspark.files import SparkFiles
 from pyspark.java_gateway import launch_gateway
 from pyspark.serializers import PickleSerializer, BatchedSerializer, UTF8Deserializer
-from pyspark.storagelevel import StorageLevel
+from pyspark.storagelevel import *
 from pyspark.rdd import RDD
 from pyspark.context import SparkContext
 
@@ -83,26 +83,9 @@ def awaitTermination(self, timeout=None):
         else:
             self._jssc.awaitTermination()
 
-    def checkpoint(self, directory):
-        raise NotImplementedError
-
-    def fileStream(self, directory, filter=None, newFilesOnly=None):
-        raise NotImplementedError
-
-    def networkStream(self, receiver):
-        raise NotImplementedError
-
-    def queueStream(self, queue, oneAtATime=True, defaultRDD=None):
-        raise NotImplementedError
-
-    def rawSocketStream(self, hostname, port, storagelevel):
-        raise NotImplementedError
-
-    def remember(self, duration):
-        raise NotImplementedError
-
-    def socketStream(hostname, port, converter,storageLevel):
-        raise NotImplementedError
+    # start from simple one. storageLevel is not passed for now.
+    def socketTextStream(self, hostname, port):
+        return DStream(self._jssc.socketTextStream(hostname, port), self, UTF8Deserializer())
 
     def start(self):
         self._jssc.start()