Notes

import sys
import re

from pyspark import SparkContext

def flatMapFunc(document):
"""
Takes in document, which is a key, value pair, where document[0] is the
document ID and document[1] is the contents of the document. We want
a distinct list of all the words in each document to pass in to our map
function, but if we call the transformation distinct() on all the words from
all the documents with the current code, we won't get quite what we want. How
can we use the document ID to help get around that problem?
"""
""" Your code here. """
lst = re.findall(r"w+", document[1]) #finds *all* the matches and returns them as a list of strings

# we get unique words in the list
return list(set(lst))

def mapFunc(arg):
""" Your code here. """
# we create the key value pairs --> (word, count)
return (arg, 1)

def reduceFunc(arg1, arg2):
""" Your code here. """
# we add all the values from pairs whose key is the same
return arg1 + arg2

def docwordcount(file_name, output="spark-wc-out-docwordcount"):
sc = SparkContext("local[8]", "DocWordCount")
file = sc.sequenceFile(file_name)

counts = file.flatMap(flatMapFunc)
.map(mapFunc)
.reduceByKey(reduceFunc)
.sortByKey() #counts.sortByKey()

counts.coalesce(1).saveAsTextFile(output)

""" Do not worry about this """
if __name__ == "__main__":
argv = sys.argv
if len(argv) == 2:
docwordcount(argv[1])
else:
docwordcount(argv[1], argv[2])


_______________________________________________________
import sys
import re

from pyspark import SparkContext

def flatMapFunc(document):
"""
Takes in document, which is a key, value pair, where document[0] is the
document ID and document[1] is the contents of the document.
HINT: You need to keep track of three things, word, document ID, and the
index inside of the document, but you are working with key, value pairs.
Is there a way to combine these three things and make a key, value pair?
"""
""" Your code here. """
lst = re.findall(r"w+", document[1])

# sets are unordered but makes the elems unique
unique_lst = set(lst)
final_lst = []

# loop thru unique_lst
for word in unique_lst:
count = 0;
id, index = "", ""
for elem in lst:
if elem == word:
id = (word + " " + str(document[0]))
index = str(count)

#make a id and index pair for key and value
final_lst.append((id, index))
count+=1
return final_lst



def mapFunc(arg):
""" Your code here. """
#take the pair from flatMapFunc, and return the pair
return (arg[0], arg[1])

def reduceFunc(arg1, arg2):
""" Your code here. """
# get all the indices, if the key is the same (id)
return arg1 + " " + arg2

def index(file_name, output="spark-wc-out-index"):
sc = SparkContext("local[8]", "Index")
file = sc.sequenceFile(file_name)

indices = file.flatMap(flatMapFunc)
.map(mapFunc)
.reduceByKey(reduceFunc)
.sortByKey()

indices.coalesce(1).saveAsTextFile(output)

""" Do not worry about this """
if __name__ == "__main__":
argv = sys.argv
if len(argv) == 2:
index(argv[1])
else:
index(argv[1], argv[2])