ó ċ#Yc@såddlmZddlZddlmZddlmZddlZddlmZddl m Z ddl m Z ddl m Z dd l mZdd lTd dd „ƒYZd efd„ƒYZdefd„ƒYZdS(iÿÿÿÿ(tprint_functionN(tConfig(tInsecureClient(tadd(t SparkConf(tdatetime(t SparkSession(t SparkContext(t*tSparkcBseZd„Zd„ZRS(cCsd|_d|_dS(N(tNonetsctsorted_user_record_list(tself((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyt__init__s cCsœtƒ}|jdƒ|jdƒ|jddƒ|jddƒ|jddƒtd |ƒ}|jd ƒ|jd ƒ|jd ƒ|jd ƒ|jdƒ|jdƒ||_dS(Nsspark://namenode:7077soff-peaksspark.executor.memoryt128gsspark.executor.corest16sspark.scheduler.modetFAIRtconfsModule/Shapely-1.6b4.zipsexploring/Spark.pysexploring/Transportation.pysexploring/Record.pysexploring/TransRegions.pysexploring/__init__.py(Rt setMastert setAppNametsetRt addPyFiletaddFileR (R RR ((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pytconnects         (t__name__t __module__RR(((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyR s t SubwaySparkcBsweZd„Zd„Zd„Zd„Zd„Zd„Zd„Zd„Z d„Z d „Z e d „Z d „ZRS( cCsPd|_d|_tƒ|_d|_d|_d|_d|_|j ƒdS(N( R t file_patht input_datatSubwaytsubwaytsorted_record_listtuser_trip_pairttrip_average_timetin_vehicle_timeR(R ((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyR"s       cCs|jj|ƒ|_dS(N(R t parallelizeR(R t inputFile((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pytsetLocalInputFile+scCs ||_dS(N(R(R R((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyt setInputData-scCs#||_|jt|jƒƒdS(N(RR&topen(R R((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pytsetLocalFilePath/s cCs/||_|j|jj|jƒjƒƒdS(N(RR'R ttextFiletcache(R R((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pytsetHDFSFilePath2s cCsp|j}|j|jjƒjd„ƒ}|jd„ƒjƒ}|jtƒjd„ƒjd„ƒ|_dS(s^ build the record list for each user and sort it by the time :return: cSs |dk S(N(R (tx((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyt<scSs |j|fS(N(tuser_id(R-((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyR.=scSs|\}}t|ƒdkS(Ni(tlen(t.0tktv((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyR.>scSs%|\}}|t|dd„ƒfS(NtkeycSs|jS(N(ttime(trecord((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyR.>s(tsorted(R1R2R3((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyR.>sN( RtmapRt subwayOneRowtfiltert groupByKeyt mapValuestlistR (R tone_filetsubway_record_listtuser_record_list((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pytbuildRecordList6s !cCs+|jj|jjƒjd„ƒ|_dS(NcSs|\}}t|ƒdkS(Ni(R0(R1R2R3((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyR.As(R R8RtsortedRecordsToTripR:R!(R ((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pyt buildTripList@scCs4|jj|jjƒ}|j|jjƒ|_dS(N(R!tflatMapRtmapToStationTimeMappingt reduceByKeyt minimumByKeyR#(R ttrip_time_pair((sA/home/zf72/Dropbox/projects/off-peak-trans/src/exploring/Spark.pytbuildInVehicleTimeCscCs²|jj|jjƒj|jƒ}|j|jjƒjd„ƒ}|jd„d„d„ƒjd„ƒ|_ i}x)|j j ƒD]}|d||ds   P