diff --git a/examples/src/main/python/streaming/network_wordcount.py b/examples/src/main/python/streaming/network_wordcount.py index f242f8d29658a..cd2a8a73de63b 100644 --- a/examples/src/main/python/streaming/network_wordcount.py +++ b/examples/src/main/python/streaming/network_wordcount.py @@ -1,5 +1,4 @@ import sys -from operator import add from pyspark.streaming.context import StreamingContext from pyspark.streaming.duration import * @@ -12,10 +11,10 @@ duration=Seconds(1)) lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2])) - words = lines.flatMap(lambda line: line.split(" ")) - mapped_words = words.map(lambda word: (word, 1)) - count = mapped_words.reduceByKey(add) - count.pyprint() + counts = lines.flatMap(lambda line: line.split(" "))\ + .map(lambda word: (word, 1))\ + .reduceByKey(lambda a,b: a+b) + counts.pyprint() ssc.start() ssc.awaitTermination() diff --git a/examples/src/main/python/streaming/wordcount.py b/examples/src/main/python/streaming/wordcount.py index e0600bbdb34d2..4c62835ed8025 100644 --- a/examples/src/main/python/streaming/wordcount.py +++ b/examples/src/main/python/streaming/wordcount.py @@ -8,13 +8,14 @@ print >> sys.stderr, "Usage: wordcount " exit(-1) - ssc = StreamingContext(appName="PythonStreamingWordCount", duration=Seconds(1)) + ssc = StreamingContext(appName="PythonStreamingWordCount", + duration=Seconds(1)) lines = ssc.textFileStream(sys.argv[1]) - words = lines.flatMap(lambda line: line.split(" ")) - mapped_words = words.map(lambda x: (x, 1)) - count = mapped_words.reduceByKey(lambda a, b: a+b) - count.pyprint() + counts = lines.flatMap(lambda line: line.split(" "))\ + .map(lambda x: (x, 1))\ + .reduceByKey(lambda a, b: a+b) + counts.pyprint() ssc.start() ssc.awaitTermination()