PySpark

def loadBind(path,name):
    #load the raw dataSet to Spark dataframe
    df = spark.read.csv(path,nullValue="NA",inferSchema="true", header="true") 
    
    #bind the Spark dataframe to a name, which can be referred to in SQL
    #Notice gotcha, it's the Spark dataframe that binds the name
    df.createOrReplaceTempView(name) 
    
    return df
loadDataAndRegister = lambda x: loadBind(x,airTraffic)

# example print
data = loadDataAndRegister(sampleFile)
data.show(2) 
print("schema")
#carrierData = loadBind(sampleFile,carriers)
#print(carrierData.schema)
#spark.read.table(airports)

'''loadDataAndRegister tests'''

df = loadBind(testFile,airTraffic)

# Table "airtraffic" should exists
assert spark.sql("SHOW TABLES Like 'airtraffic'").count() == 1, "there was expected to be a table called 'airtraffic'"

# Columns should have correct values
third = df.collect()[2]
correctRow = Row(Year=2008, Month=5, DayofMonth=6, DayOfWeek=2, DepTime=611,
                             CRSDepTime=615, ArrTime=729, CRSArrTime=735, UniqueCarrier='EV',
                             FlightNum=4794, TailNum='N916EV', ActualElapsedTime=78,
                             CRSElapsedTime=80, AirTime=58, ArrDelay=-6, DepDelay=-4,
                             Origin='ROA', Dest='ATL', Distance=357, TaxiIn=9, TaxiOut=11,
                             Cancelled=0, CancellationCode=None, Diverted=0, CarrierDelay=None,
                             WeatherDelay=None, NASDelay=None, SecurityDelay=None,
                             LateAircraftDelay=None).asDict()

assert third.asDict() == correctRow, "the row was expected to be %s but it was %s" % (correctRow, third.asDict())

def CountFlight(df):
    # YOUR CODE HERE
    CountDesc = (df
                 .groupby("TailNum")
                 .count()
                 .sort(f.desc("count")) 
                )
    #groupby() aggregates dataframe to groups by TailNum
    #count() counts elements in each group
    #sort(f.desc("count")) sorts by the count 
    
    CountDescExNull = (CountDesc
                       .filter(CountDesc.TailNum.isNotNull())
                      )
    #filter() will only include non-Null TailNum
    
    return CountDescExNull

# example print
data = loadBind(testFile,airTraffic)
CountFlight(data).show(20)
#flightCount(data)[:5]

'''flightCount tests'''

data = loadBind(testFile,airTraffic)
        
correct = [Row(TailNum='N824AS', count=2),
           Row(TailNum='N856AS', count=1),
           Row(TailNum='N886AS', count=1),
           Row(TailNum='N916EV', count=1),
           Row(TailNum='N873AS', count=1),
           Row(TailNum='N881AS', count=1)]

correctRows(CountFlight(data).collect(), correct)

def SecurityCancel(df):
    output= spark.sql("""SELECT FlightNum, Dest from airtraffic WHERE CancellationCode='D';""")
    return output

# example print
data = loadBind(sampleFile,airTraffic)
SecurityCancel(data).show(5)

'''cancelledDueToSecurity tests'''

data = loadDataAndRegister(testFile)
correct = [Row(FlightNum=4794, Dest='JFK'), Row(FlightNum=4794, Dest='ATL')]

correctRows(SecurityCancel(data).collect(), correct)

def MaxWeatherDelay(df):
    output = spark.sql(""" SELECT MAX(WeatherDelay) as _c0 from airtraffic WHERE Month >= 1 and Month <= 3""")
    return output
    

# example print
data = loadBind(sampleFile,airTraffic)
MaxWeatherDelay(data).show()

delay = longestWeatherDelay(data).first()[0]
print(f"longest weather delay between jan and mar is {delay}"   )

'''longestWeatherDelay tests'''

data = loadDataAndRegister(testFile)
test = longestWeatherDelay(data).first()[0]

assert test == 7, "the longest weather delay was expected to be 7 but it was %s" % test

def didNotFly(df):
    #a = spark.sql("SELECT DESCRIPTION from AirTraffic INNER JOIN carriers on carriers.Code = AirTraffic.UniqueCarrier;")
    a = spark.sql("SELECT Description FROM carriers WHERE Code NOT IN (SELECT Code from airtraffic INNER JOIN carriers on carriers.Code = airtraffic.UniqueCarrier);")
    #a = spark.sql("Select Code FROM carrierse")
    return a

data = loadBind(testFile,airTraffic)
didNotFly(data).show(5)

'''didNotFly tests'''

data = loadDataAndRegister(testFile)
test = didNotFly(data).count()

assert test == 1489, "the amount of airlines that didn't fly was expected to be 1489 but it was %s" % test

def flightsFromVegasToJFK(df):
    fdata = CountFlight(df)
    #loadDataAndRegister("flightCount")
    fdata.createOrReplaceTempView("flightCount")
    output = spark.sql("""
    SELECT Description, count as Num FROM 
        flightCount 
        INNER JOIN
        (SELECT * FROM 
            carriers 
            INNER JOIN 
            (SELECT DISTINCT UniqueCarrier,TailNum FROM airtraffic WHERE Origin = 'LAS' AND Dest = 'JFK') as T1
            ON
            carriers.Code = T1.UniqueCarrier) as T2
        ON flightCount.TailNum = T2.TailNum
            ORDER BY Num DESC
    
    """)
    # YOUR CODE HERE
    return output

# example print
data = loadBind(testFile,airTraffic)
flightsFromVegasToJFK(data).show(5)

'''flightsFromVegasToJFK tests'''

data = loadDataAndRegister(testFile)
correct = [Row(Description='Titan Airways', Num=1),
           Row(Description='Atlantic Southeast Airlines', Num=1)]
correctRows(flightsFromVegasToJFK(data).collect(), correct)

def TaxiTime(df):
    output = spark.sql("""
        SELECT T1.Origin as airport,(A1+A2)/2 as taxi FROM
            (SELECT Origin,AVG(TaxiIn) AS A1 FROM AirTraffic GROUP BY Origin) AS T1
            INNER JOIN
            (SELECT Dest,AVG(TaxiOut) AS A2 FROM AirTraffic GROUP BY Dest) AS T2
            ON
                T1.Origin = T2.Dest
            ORDER BY
                taxi ASC
            
    """)
    
    
    return output

data = loadBind(testFile,airTraffic)
TaxiTime(data).show(5)

'''timeSpentTaxiing tests'''

data = loadDataAndRegister(testFile)
correct = [Row(airport='LAS', taxi=11.0), Row(airport='JFK', taxi=13.25)]
correctRows(TaxiTime(data).collect(), correct)

def distanceMedian(df):
    a = spark.sql("""
        SELECT percentile(T1,0.5)as `_ c0` FROM
            (SELECT Distance FROM AirTraffic) AS Tab(T1)
    """)
    # YOUR CODE HERE
    return a

data = loadBind(testFile,airTraffic)
distanceMedian(data).show()

'''distanceMedian tests'''

data = loadDataAndRegister(testFile)
test = distanceMedian(data).first()[0]
assert test == 357.0, "the distance median was expected to be 357.0 but it was %s" % test

def score95(df):
    a = spark.sql("""
        SELECT percentile(T1,0.95)as `_ c0` FROM
            (SELECT CarrierDelay FROM AirTraffic) AS Tab(T1)
    """)
    # YOUR CODE HERE
    return a

data = loadBind(testFile,airTraffic)
score95(data).show()

'''score95 tests'''

data = loadDataAndRegister(testFile)
test = score95(data).first()[0]
assert test == 17.0, "the score95 was expected to be 17.0 but it was %s" % test

def cancelledFlights(df):
    a = spark.sql("""
        SELECT airports.airport, airports.city, T1.percentage 
        FROM
            (SELECT Origin,SUM(Cancelled)/COUNT(Origin) as percentage FROM AirTraffic 
            GROUP BY Origin) AS T1
        INNER JOIN
            airports
        ON
            T1.Origin = airports.iata
        ORDER BY
            T1.percentage DESC,
            airports.airport DESC
    """)
    
    # YOUR CODE HERE
    return a

data = loadBind(testFile,airTraffic)
cancelledFlights(data).show(5)

'''cancelledFlights tests'''

data = loadDataAndRegister(testFile)
correct = [Row(airport='McCarran International', city='Las Vegas', percentage=0.5),
           Row(airport='Roanoke Regional/ Woodrum ', city='Roanoke', percentage=0.25)]
correctRows(cancelledFlights(data).collect(), correct)

def leastSquares(df):
    try:
        a = spark.sql("""
            SELECT DD, WD, DD*WD as Prod, DD*DD as Sqx  
            FROM
            (SELECT DepDelay as DD,AVG(WeatherDelay) as WD FROM AirTraffic
            WHERE DepDelay > 0
            GROUP BY DepDelay) AS T1
        """)
        # YOUR CODE HERE
        xsum = a.groupBy().sum('DD').collect()[0][0]
        ysum = a.groupBy().sum('WD').collect()[0][0]
        psum = a.groupBy().sum('Prod').collect()[0][0]
        sqxsum = a.groupBy().sum('Sqx').collect()[0][0]
        #print(xsum)
        N = a.count()
        varM = ((N * psum)-(xsum*ysum))/((N * sqxsum)-(xsum**2))
        varB = ((ysum*sqxsum)-(xsum*psum))/((N*sqxsum)-(xsum**2)) 
        return varB,varM
    except error:
        return 0.0,0.0

data = loadBind(testFile,airTraffic)
leastSquares(data)
#leastSquares(data)[0].show(4)
#print(leastSquares(data)[1])
#print(leastSquares(data)[2])

'''leastSquaresTests'''

data = loadDataAndRegister(testFile)
test = leastSquares(data)
assert test == (952.0, -56.0), "the answer was expected to be (952.0, -56.0) but it was %s" % test

spark.stop()