er.py

import mozillametricstools.common.functions as cf
from  mozillametricstools.common.functions import dateRangeFromTo
import datetime, time

frame =  sqlContext.read.load(cf.latest_longitudinal_path())
first = frame.limit(5).rdd.collect()



def dateDiff(d,delta=0):
    x = datetime.datetime.strptime(d, "%Y-%m-%d").date()
    return (d,x, (x +  datetime.timedelta(days=delta)).strftime("%Y-%m-%d"))

    
def smry(p):
    submissiondates = [ x[0:10] for x in p.submission_date ]
    ssubmissiondates = set(submissiondates)
    activity = []
    for g in analysis_dates:
        thisdateNow,thisdate28,thisdate7, thisdate1 = g['thisdateNow'],g['thisdate28'],g['thisdate7'],g['thisdate1']

        s = ( (thisdate1,'totalActiveDays28') ,N* sum([ x >= thisdate28 and x < thisdateNow for x in   ssubmissiondates]))
        yield s
        yield ((thisdate1, 'wasActive28'), N*1*(s[1]>0))
        
        s = ( (thisdate1,'totalActiveDays7') , N*sum([ x >= thisdate7 and x < thisdateNow for x in   ssubmissiondates]))
        yield s 
        yield ((thisdate1, 'wasActive7'), N*1*(s[1]>0))

        s = ( (thisdate1,'totalActiveDays1') , N*sum([ x >= thisdate1 and x < thisdateNow for x in   ssubmissiondates]))
        yield s


        
def makeDateDict(x):
    a = {}
    a['thisdateNow'] =  x
    a['thisdate28'] =  dateDiff(x, delta=-28)[2]
    a['thisdate7'] =  dateDiff(x, delta=-7)[2]
    a['thisdate1'] =  dateDiff(x, delta=-1)[2]
    return a
    
analysis_dates = [ makeDateDict(x) for x in dateRangeFromTo(start='2016-04-18', length=90, by=-3)]
N = 100
sf = frame.filter("normalized_channel = 'release'").select("submission_date")
sf3 = sf.flatMap(smry)
sf4 = sf3.reduceByKey(lambda a,b: a+b, 200)
sf4.saveAsPickleFile("{}erdata".format(sguhafolder))