############# ############# ############# ############# #############
# filterData
# by JAG3
#
############# ############# ############# ############# #############

from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext, Row
from pyspark.sql.types import BooleanType
from datetime import date
import sys
import argparse
sys.path.insert(0, './lib/')
from to_parquet import csvToDataFrame
import fspLib
import shapeReader

# HARD CODE YOU INPUT DATA SETS AND DATA TYPES
DATA_SETS = {"/data/ingest/twitter/success/":2}


LOWER_TIME = date(2006,03,21)
UPPER_TIME = date(3000,01,01)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("shapeFile", help="The shape file path")
    parser.add_argument("outputPath",help="Output destination")
    parser.add_argument("-jobNm", help="Application name, default = 'Geqe Data Filter'",default='Geqe data filter.')
    parser.add_argument("-cNum", type=int, help="Number of processes to coalesce initial input data to, default = 3",default = 8)
    parser.add_argument("--stopWordsFile",help="File path to a stop words list. One word per line. default=inputFiles/stopWordList.txt",default="inputFiles/stopWordList.txt")
    parser.add_argument("-sCustStop", help="Comma seperated list of stop words to add include on this run",default='')
    args = parser.parse_args()

    shapeFile = args.shapeFile
    outputPath = args.outputPath
    jobNm = args.jobNm
    cNum = args.cNum
    stopWordsPath = args.stopWordsFile
    sCustStop = args.sCustStop

    #Declare Spark Context
    conf = SparkConf().setAppName(jobNm)
    sc = SparkContext(conf = conf)
    sqlContext = SQLContext(sc)

    #Read in stop word list early to get notified of issues early in process
    bc_lStopWords = fspLib.load_stopwords(sc,stopWordsPath,sCustStop)

    #Create polygon list and broadcast variable based on it
    lPolygon = shapeReader.readInShapeJson(shapeFile)
    bc_lTargetPolygons = sc.broadcast(lPolygon)

    # register SQL functions
    sqlContext.registerFunction("hasScorableWord", lambda text: fspLib.hasScorableWord(text,True,bc_lStopWords),returnType=BooleanType())
    sqlContext.registerFunction("inRegionOfInterest", lambda lat,lon: fspLib.inROI(lat,lon,bc_lTargetPolygons),returnType=BooleanType())

    # read in each data set and filter for region / time of interest
    dfs = []
    i = 0
    for dataSet,dataType in DATA_SETS.iteritems():
        print 'processing data set: ',dataSet
        i+=1
        tablename = 'table'+str(i)
        df = csvToDataFrame(sc,sqlContext,dataSet,dataType)
        df.registerTempTable(tablename)
        sqlCommand = "SELECT * from %s WHERE inRegionOfInterest(lat,lon) AND hasScorableWord(text) " % (tablename)
        #sqlCommand = "SELECT * from %s WHERE hasScorableWord(text) " % (tablename)
        df = sqlContext.sql(sqlCommand)
        dfs.append(df)

    # union ll dfs together
    print 'performing union on all of: ',dfs
    unionDF = dfs[0]
    for df in dfs[1:]:
        print 'performing df union'
        unionDF = unionDF.unionAll(df)
    unionDF.coalesce(cNum).saveAsParquetFile(outputPath)

@@ -3119,25 +3119,21 @@
 um).
-saveAsP
+write.p
 arquet
-File
 (out

update depreciated method