<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import csv <EOL> import math <EOL> from functools import partial <EOL> from StringIO import StringIO <EOL> from pyspark import SparkConf , SparkContext <EOL> APP_NAME = "<STR_LIT>" <EOL> DATASET = os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT:..>" , "<STR_LIT:data>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> def counters ( line ) : <EOL> """<STR_LIT>""" <EOL> reader = csv . reader ( StringIO ( line ) , delimiter = '<STR_LIT:U+002C>' ) <EOL> row = reader . next ( ) <EOL> airport = row [ <NUM_LIT:4> ] <EOL> try : <EOL> delay = float ( row [ <NUM_LIT:8> ] ) <EOL> except ValueError : <EOL> delay = <NUM_LIT:0.0> <EOL> return ( airport , ( <NUM_LIT:1> , delay , delay ** <NUM_LIT:2> , delay , delay ) ) <EOL> def aggregation ( item1 , item2 ) : <EOL> """<STR_LIT>""" <EOL> count1 , total1 , squares1 , min1 , max1 = item1 <EOL> count2 , total2 , squares2 , min2 , max2 = item2 <EOL> minimum = min ( ( min1 , min2 ) ) <EOL> maximum = max ( ( max1 , max2 ) ) <EOL> count = count1 + count2 <EOL> total = total1 + total2 <EOL> squares = squares1 + squares2 <EOL> return ( count , total , squares , minimum , maximum ) <EOL> def summary ( aggregate ) : <EOL> """<STR_LIT>""" <EOL> ( airport , ( count , total , square , minimum , maximum ) ) = aggregate <EOL> try : <EOL> mean = total / float ( count ) <EOL> stddev = math . sqrt ( ( square - ( total ** <NUM_LIT:2> ) / count ) / count - <NUM_LIT:1> ) <EOL> return ( airport , ( count , mean , stddev , minimum , maximum ) ) <EOL> except Exception : <EOL> return ( airport , ( count , None , None , minimum , maximum ) ) <EOL> def main ( sc ) : <EOL> """<STR_LIT>""" <EOL> delays = sc . textFile ( DATASET ) . map ( counters ) <EOL> delays = delays . reduceByKey ( aggregation ) <EOL> delays = delays . map ( summary ) <EOL> delays . saveAsTextFile ( "<STR_LIT>" ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> conf = SparkConf ( ) . setAppName ( APP_NAME ) <EOL> sc = SparkContext ( conf = conf ) <EOL> main ( sc ) </s>
