Skip to content
Snippets Groups Projects
Select Git revision
  • master default
1 result
Blame Permalink
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
grading.py 2.79 KiB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84 






from pyspark import SparkContext
from functions import *
import re
import sys

sc = SparkContext("local", "Simple App")
setDefaultAnswer(sc.parallelize([0]))

taskno = int(sys.argv[1])

if taskno == 1 or taskno == -1:
	playRDD = sc.textFile("datafiles/play.txt")
	### Task 1
	print "=========================== Task 1"
	task1_result = task1(playRDD)
	for x in task1_result.takeOrdered(100):
		print x

if taskno == 2 or taskno == -1:
	### Task 2
	print "=========================== Task 2"
	nobelRDD = sc.textFile("datafiles/prize.json")
	task2_result = nobelRDD.map(json.loads).flatMap(task2_flatmap).distinct()
	for x in task2_result.takeOrdered(100):
		print x.encode('utf-8')

if taskno == 3 or taskno == -1:
	#### Task 3
	print "=========================== Task 3"
	nobelRDD = sc.textFile("datafiles/prize.json")
	task3_result = task3(nobelRDD)
	for x in task3_result.takeOrdered(100):
		print x

if taskno == 4 or taskno == -1:
	#### Task 4
	logsRDD = sc.textFile("datafiles/NASA_logs_sample.txt")
	print "=========================== Task 4"
	task4_result = task4(logsRDD, ['02/Jul/1995', '03/Jul/1995', '04/Jul/1995', '05/Jul/1995', '06/Jul/1995'])
	for x in task4_result.takeOrdered(100):
		print x.encode('utf-8')

if taskno == 5 or taskno == -1:
	#### Task 5
	print "=========================== Task 5"
	amazonInputRDD = sc.textFile("datafiles/amazon-ratings.txt")
	amazonBipartiteRDD = amazonInputRDD.map(lambda x: x.split(" ")).map(lambda x: (x[0], x[1])).distinct()
	task5_result = task5(amazonBipartiteRDD)
	for x in task5_result.takeOrdered(100):
		print x

if taskno == 6 or taskno == -1:
	#### Task 6
	print "=========================== Task 6"
	logsRDD = sc.textFile("datafiles/NASA_logs_sample.txt")
	task6_result = task6(logsRDD, '03/Jul/1995', '05/Jul/1995')
	for x in task6_result.takeOrdered(100):
		print x

if taskno == 7 or taskno == -1:
	#### Task 7
	print "=========================== Task 7"
	nobelRDD = sc.textFile("datafiles/prize.json")
	task7_result = task7(nobelRDD)
	for x in task7_result.takeOrdered(100):
		print x

if taskno == 8 or taskno == -1:
	#### Task 8 -- we will start with a non-empty currentMatching and do a few iterations
	amazonInputRDD = sc.textFile("datafiles/amazon-ratings.txt")
	amazonBipartiteRDD = amazonInputRDD.map(lambda x: x.split(" ")).map(lambda x: (x[0], x[1])).distinct()
	print "=========================== Task 8"
	currentMatching = sc.parallelize([('user1', 'product8')])
	res1 = task8(amazonBipartiteRDD, currentMatching)
	print "Found {} edges to add to the matching".format(res1.count())
	for x in res1.takeOrdered(100):
		print x
	currentMatching = currentMatching.union(res1)
	res2 = task8(amazonBipartiteRDD, currentMatching)
	print "Found {} edges to add to the matching".format(res2.count())
	for x in res2.takeOrdered(100):
		print x
	currentMatching = currentMatching.union(res2)