Modify the PageRank Algorithm and run as stream application

Modify the application from the previous exercise so it reads all the files in the directory and (2) captures any new records/files as a streaming application.

• Open the terminal and start the the streaming application
• Open a new terminal and upload the files listed in this exercise using the 'copyFromLocal' command. Allow some intervals (for example 1-2 minutes) between the executions of the copyFromLocal command while observing the output of the other terminal's data streaming application. Notice how adding more files will change the rankings of the accounts.

My codes run as streaming (on AWS emr using putty), it is not adding the CSV I am adding to the 'new' folder. not providing a new rank.

pagerank alogrithm

import sys
from pyspark.sql import SparkSession

if __name__ == "__main__":
        if len(sys.argv) < 4:
                print >> sys.stderr
                "Usage: PageRank.py <input-file> <output-file> <iterations>"
                sys.exit()
        spark = SparkSession.builder.getOrCreate()
        spark.sparkContext.setLogLevel("WARN")

        def computeContribs(neighbors, rank):
                for neighbor in neighbors:
                        yield(neighbor, rank/len(neighbors))
        links = spark.sparkContext.textFile(sys.argv[1]).\
                map(lambda line: line.split(',')).\
                map(lambda pages: (pages[0], pages[1])).\
                distinct().\
                groupByKey().\
                map(lambda x: (x[0], list(x[1])))
        ranks = links.map(lambda element: (element[0], 1.0))
        iterations = int(sys.argv[3])
        for x in range(iterations):
                contribs = links.join(ranks).flatMap(lambda row:   computeContribs(row[1][0], row[1][1]))
                print("\n")
                print("------- Iter: " + str(x) + " --------")
                ranks = contribs.reduceByKey(lambda v1, v2: v1+v2).map(lambda nr: (nr[0], nr[1] * 0.85 + 0.15))
                for rank in ranks.collect(): print(rank)
                print("\n")
        print("------- Final Results --------")
        for rank in ranks.collect(): print(rank)
        ranks.saveAsTextFile(sys.argv[2])
        spark.stop()

mycode import sys

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
if __name__ == "__main__":
        sc = SparkContext(appName="Rank")
        ssc = StreamingContext(sc, 10)
        ssc.sparkContext.setLogLevel("WARN")
        def computeContribs(neighbors, rank):
                for neighbor in neighbors:
                        yield(neighbor, rank/len(neighbors))
        links = ssc.textFileStream('new/').\
                map(lambda line: line.split(',')).\
                map(lambda pages: (pages[0], pages[1])).\
                groupByKey().\
                map(lambda x: (x[0], list(x[1])))
        ranks = links.map(lambda element: (element[0], 1.0))
        contribs = links.join(ranks).flatMap(lambda row:   computeContribs(row[1][0], row[1][1]))
        ranks = contribs.reduceByKey(lambda v1, v2: v1+v2).map(lambda nr: (nr[0], nr[1] * 0.85 + 0.15))
        ranks.pprint()
        ssc.start()
        ssc.awaitTermination()