from pyspark import SparkConf, SparkContext
import math

conf = SparkConf().setMaster("local").setAppName("app")
sc = SparkContext(conf = conf)

linesPrvaGodinaRDD = sc.textFile("flcas1.txt")
linesPetaGodinaRDD = sc.textFile("flcas5.txt")
prvaGodina = linesPrvaGodinaRDD.map(lambda line: line.split())
petaGodina = linesPetaGodinaRDD.map(lambda line: line.split())

# di razlika prvog i drugog merenja
# di nadvuceno razlika pa arsr
i=1

def funkcija(x):
    global i
    i=i+1
    return (i-1,int(x[3]))


prvaGodina = prvaGodina.map(funkcija)
i=1
petaGodina = petaGodina.map(funkcija)

razlika = prvaGodina.join(petaGodina).map(lambda x: (1,x[1][0]-x[1][1]))

suma = razlika.reduceByKey(lambda x,y: x+y).values().first()
count = prvaGodina.count()

arSr = suma*1.0/count

razlikeNaKvadrat = razlika.map(lambda x: (1,(x[1]-arSr)*(x[1]-arSr)))

sumaRazlikeNaKvadrat = razlikeNaKvadrat.reduceByKey(lambda x,y:x+y).values().first()

stDev = math.sqrt(sumaRazlikeNaKvadrat/(count-1))

t = arSr*1.0/(stDev*1.0/(math.sqrt(count)))

f = open("tSpark.txt","w")
f.write(str(t))
f.close()

#print("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",arSr)
#razlika.saveAsTextFile("niz")

#2.

prvaGodina = linesPrvaGodinaRDD.map(lambda line: line.split())
petaGodina = linesPetaGodinaRDD.map(lambda line: line.split())

i=1
def funkcija2(x):
    global i
    i=i+1
    return (i-1,(str(x[1]),int(x[2]),int(x[3])))

#prva = prvaGodina.map(lambda (god,pol,brGod,skor) : (1,(str(pol),int(brGod),int(skor)))).filter(lambda x: x[1][0] == 'Z')
#peta = petaGodina.map(lambda (god,pol,brGod,skor) : (1,(str(pol),int(brGod),int(skor)))).filter(lambda x: x[1][0] == 'Z')

prva = prvaGodina.map(funkcija2).filter(lambda x: x[1][0] == 'Z')
peta = petaGodina.map(funkcija2).filter(lambda x: x[1][0] == 'Z')

razlika = prva.join(peta).map(lambda x: (x[0],(x[1][0][2]-x[1][1][2],x[1][0][0],x[1][0][1])))

maksSmanjenje = razlika.map(lambda x: (x[1][0],(x[0],x[1][1],x[1][2]))).sortByKey(ascending=False).keys().first()

resenje = razlika.filter(lambda x: x[1][0] == maksSmanjenje)

pom = prva.join(resenje).map(lambda x: (1,(x[1][0][0],x[1][0][1],x[1][0][2])))

f = open("najStudentSpark.txt","w")
for i in pom.collect():
    #f.write(str(i[0])+" "+str(i[1][0])+" "+str(i[1][1])+" "+str(i[i][2])+'\n')
    f.write(str(i[0]) +" "+ str(i[1][0]) +" "+ str(i[1][1]) +" "+ str(i[1][2]) + "\n")
f.close()


#pom.saveAsTextFile("niz")
#print("AAAAAAAAAAAAAAAAAAAAAAAAAAAAA",pom.collect())