from pyspark import SparkContext, SparkConf, StorageLevel
from pyspark.streaming import StreamingContext


checkpoint_dir = "checkpoint-online-wordcount"

def startSSC():
        conf = SparkConf().set("spark.streaming.receiver.writeAheadLog.enable" , "true")
        sc = SparkContext("local[*]", "NetworkWordCount")
        ssc = StreamingContext(sc, 1)
        ssc.checkpoint(checkpoint_dir)

        lines = ssc.socketTextStream("localhost", 9999, StorageLevel.MEMORY_ONLY)
        words = lines.flatMap(lambda line: line.split())
        pairs = words.map(lambda word: (word, 1))
        wordCounts = pairs.reduceByKey(lambda x, y: x + y)




        windowedWordCounts = pairs.\
                filter(lambda pair: pair[0].\
                startswith("#")).\
                reduceByKeyAndWindow(lambda x, y: x + y, lambda x, y: x - y, 30, 10)

        # Print the first ten elements of each RDD generated in this DStream to the console
        wordCounts.saveAsTextFiles("data/raw","text")
        windowedWordCounts.pprint(25)
        return ssc


ssc = StreamingContext.getOrCreate(checkpoint_dir, startSSC)

ssc.start()             # Start the computation
ssc.awaitTermination()  # Wait for the computation to terminate


@@ -638,18 +638,41 @@
 x + y)%0A%0A
-%0A%0A
+        #Window Operation
 %0A       
@@ -882,143 +882,139 @@
   # 
-Print the first ten elements of each RDD generated in this DStream to the console%0A        wordCounts.saveAsTextFiles(%22data/raw%22,%22text%22)
+Save the raw data in HDFS%0A        lines.saveAsTextFiles(%22data/raw%22,%22text%22)%0A%0A        # Print the values of the word counts in window
 %0A   
@@ -1048,16 +1048,17 @@
 int(25)%0A
+%0A
         
@@ -1180,16 +1180,16 @@
 utation%0A
+
 ssc.awai
@@ -1244,9 +1244,8 @@
 rminate%0A
-%0A

Update stream-02.py